當前位置 主頁 > 服務器問題 > Linux/apache問題 > 最大化 縮小

    Spring純Java配置集成kafka代碼實例

    欄目:Linux/apache問題 時間:2020-01-19 00:18

    這篇文章主要介紹了Spring純Java配置集成kafka代碼實例,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

    KafkaConfig.java

    package com.niugang.config;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.OffsetAndMetadata;
    import org.apache.kafka.clients.consumer.OffsetCommitCallback;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.common.TopicPartition;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import org.apache.kafka.common.serialization.StringSerializer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.annotation.EnableKafka;
    import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
    import org.springframework.kafka.core.ConsumerFactory;
    import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
    import org.springframework.kafka.core.DefaultKafkaProducerFactory;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.core.ProducerFactory;
    import org.springframework.kafka.listener.AbstractMessageListenerContainer;
    
    import com.niugang.controller.SenderConttoller;
    
    /**
     * 
     * @ClassName: KafkaConfig
     * @Description:kafka配置類,基于spring java純配置的
     * @author: niugang
     * @date: 2018年10月20日 下午8:04:26
     * @Copyright: 863263957@qq.com. All rights reserved.
     *
     */
    @Configuration
    @EnableKafka
    public class KafkaConfig {
    
      private Logger logger = LoggerFactory.getLogger(KafkaConfig.class);
    
      @Bean
      public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 偏移量提交方式
        // factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.COUNT);
        // 異步提交偏移量(默認就是true)
        // factory.getContainerProperties().setSyncCommits(true);
        //回調函數經常用于記錄提交錯誤
        /*factory.getContainerProperties().setCommitCallback(new OffsetCommitCallback() {
    
          @Override
          public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
            if (exception != null) {
              logger.error("Commit failed for effsets {}", offsets, exception);
            }
    
          }
        });*/
        factory.setConsumerFactory(consumerFactory());
        return factory;
      }
    
      /**
       * 消費者工廠配置
       * 
       * @return
       */
      @Bean
      public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProps());
      }
    
      /**
       * 生產者工廠配置
       * 
       * @return
       */
      @Bean
      public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(senderProps());
      }
    
      /**
       * kafka發送消息模板
       * 
       * @return
       */
      @Bean
      public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
      }
    
      /**
       * 消費者監聽
       * 
       * @return
       */
      @Bean
      public ConsumerListener listener() {
        return new ConsumerListener();
      }
    
      /**
       * 消費配置方法
       * 
       * @return
       */
      private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka_group_1");
        /**
         * enable.auto.commit 默認5秒自動提交偏移量
         */
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        /**
         * kafka是基于key-value鍵值對的,以下配置key和value的反序列化放
         */
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
      }
    
      /**
       * 生產者配置方法
       * 
       * 生產者有三個必選屬性
       * <p>
       * 1.bootstrap.servers broker地址清單,清單不要包含所有的broker地址,
       * 生產者會從給定的broker里查找到其他broker的信息。不過建議至少提供兩個broker信息,一旦 其中一個宕機,生產者仍能能夠連接到集群上。
       * </p>
       * <p>
       * 2.key.serializer broker希望接收到的消息的鍵和值都是字節數組。 生產者用對應的類把鍵對象序列化成字節數組。
       * </p>
       * <p>
       * 3.value.serializer 值得序列化方式
       * </p>
       * 
       * 
       * @return
       */
      private Map<String, Object> senderProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        /**
         * 當從broker接收到的是臨時可恢復的異常時,生產者會向broker重發消息,但是不能無限
         * 制重發,如果重發次數達到限制值,生產者將不會重試并返回錯誤。
         * 通過retries屬性設置。默認情況下生產者會在重試后等待100ms,可以通過 retries.backoff.ms屬性進行修改
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        /**
         * 在考慮完成請求之前,生產者要求leader收到的確認數量。這可以控制發送記錄的持久性。允許以下設置:
         * <ul>
         * <li>
         * <code> acks = 0 </ code>如果設置為零,則生產者將不會等待來自服務器的任何確認。該記錄將立即添加到套接字緩沖區并視為已發送。在這種情況下,無法保證服務器已收到記錄,并且
         * <code>retries </ code>配置將不會生效(因為客戶端通常不會知道任何故障)。為每條記錄返回的偏移量始終設置為-1。
         * <li> <code> acks = 1 </code>
         * 這意味著leader會將記錄寫入其本地日志,但無需等待所有follower的完全確認即可做出回應。在這種情況下,
         * 如果leader在確認記錄后立即失敗但在關注者復制之前,則記錄將丟失。
         * <li><code> acks = all </code>
         * 這意味著leader將等待完整的同步副本集以確認記錄。這保證了只要至少一個同步副本仍然存活,記錄就不會丟失。這是最強有力的保證。
         * 這相當于acks = -1設置
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        /**
         * 當有多條消息要被發送到統一分區是,生產者會把他們放到統一批里。kafka通過批次的概念來 提高吞吐量,但是也會在增加延遲。
         */
        // 以下配置當緩存數量達到16kb,就會觸發網絡請求,發送消息
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 每條消息在緩存中的最長時間,如果超過這個時間就會忽略batch.size的限制,由客戶端立即將消息發送出去
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // key的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // value序列化方式
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
      }
    }
    
    下一篇:沒有了
777亚洲人成视频免费视频