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: [e-mail]. 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;
    }
}

 

ConsumerListener.java

package com.niugang.config;

import org.springframework.kafka.annotation.KafkaListener;

/**
 * 
 * @ClassName:  ConsumerListener   
 * @Description:消費者監聽  
 * @author: niugang
 * @date:   2018年10月21日 下午2:05:21   
 * @Copyright: [e-mail]. All rights reserved. 
 *
 */
public class ConsumerListener {
    /**
     * topicPattern:支持正則表達式
     * @param foo
     */
    @KafkaListener(id = "foo", topics = "annotated1")
    public void listen1(String foo) {
        System.out.println("接收消息為："+foo);
    }
}

 

源碼：https://gitee.com/niugangxy/kafka/tree/master/kafka-spring-boot

微信公眾號

                           ​