Kafka——Spring集成Kafka

 

准備工作

• 安裝kafka+zookeeper環境

• 利用命令創建好topic

• Pom文件,引入spring-kafka jar包這里需要注意2個地方：

1. kafka-clients 包版本與服務器端kafka-clients版本保持一致（查看服務器kafka版本方法 在kafka安裝目錄下libs 中查找kafka-clients開頭的jar文件）
2. 引入的spring-kafka 版本在2.0或者2.X 時Spring版本在5.0才能支持

　　..........
   <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>2.1.8.RELEASE</version>
   </dependency>
　　..........

參考官網 http://kafka.apache.org/documentation/ 

 

XML配置方式

• 生產者

配置：

 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
     xsi:schemaLocation="http://www.springframework.org/schema/beans  
          http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
          http://www.springframework.org/schema/context  
          http://www.springframework.org/schema/context/spring-context.xsd">
     <context:property-placeholder location="classpath*:config/application.properties" />
     <!-- 定義producer的參數 -->
     <bean id="producerProperties" class="java.util.HashMap">
         <constructor-arg>
             <map>
                 <entry key="bootstrap.servers" value="${bootstrap.servers}" />
                 <entry key="group.id" value="${group.id}" />
                 <entry key="retries" value="${retries}" />
                 <entry key="batch.size" value="${batch.size}" />
                 <entry key="linger.ms" value="${linger.ms}" />
                 <entry key="buffer.memory" value="${buffer.memory}" />            
                 <entry key="acks" value="${acks}" />              
                 <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
                 <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
             </map>
         </constructor-arg>
     </bean>
 
     <!-- 創建kafkatemplate需要使用的producerfactory bean -->
     <bean id="producerFactory"
         class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
         <constructor-arg>
             <ref bean="producerProperties" />
         </constructor-arg>
     </bean>
 
     <!-- 創建kafkatemplate bean，使用的時候，只需要注入這個bean，即可使用template的send消息方法 -->
     <bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
         <constructor-arg ref="producerFactory" />
         <constructor-arg name="autoFlush" value="true" />
         <property name="defaultTopic" value="default" />
     </bean>
 </beans>

如上圖，xml主要配置了KafkaTemplate的構造參數producerFactory和autoFlush，對應了一個KafkaTemplate源碼中的2參構造函數。

1. producerProperties：設置生產者工廠需要的配置
2. producerFactory：定義了生產者工廠構造方法
3. kafkaTemplate：定義了使用producerFactory和是否自動刷新，2個參數來構造kafka生產者模板類

發送消息：

ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("topic", "partition","key","data");

//發送成功回調
SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() {
     @Override
     public void onSuccess(SendResult<String, String> result) {
        //成功業務邏輯
     }
}

//發送失敗回調
FailureCallback failureCallback = new FailureCallback() {
     @Override
     public void onFailure(Throwable ex) {
         //失敗業務邏輯
     }
}
listenableFuture.addCallback(successCallback, failureCallback);

• 消費者

配置：

 <!-- 1.定義consumer的參數 -->
     <bean id="consumerProperties" class="java.util.HashMap">
         <constructor-arg>
             <map>
                 <entry key="bootstrap.servers" value="${bootstrap.servers}" />
                 <entry key="group.id" value="${group.id}" />
                 <entry key="enable.auto.commit" value="${enable.auto.commit}" />
                 <entry key="session.timeout.ms" value="${session.timeout.ms}" />
                 <entry key="key.deserializer"
                     value="org.apache.kafka.common.serialization.StringDeserializer" />
                 <entry key="value.deserializer"
                     value="org.apache.kafka.common.serialization.StringDeserializer" />
             </map>
         </constructor-arg>
     </bean>
 
     <!-- 2.創建consumerFactory bean -->
     <bean id="consumerFactory"
         class="org.springframework.kafka.core.DefaultKafkaConsumerFactory" >
         <constructor-arg>
             <ref bean="consumerProperties" />
         </constructor-arg>
     </bean>
 
     <!-- 3.定義消費實現類 -->
     <bean id="kafkaConsumerService" class="xxx.service.impl.KafkaConsumerSerivceImpl" />
 
     <!-- 4.消費者容器配置信息 -->
     <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
         <!-- topic -->
         <constructor-arg name="topics">
             <list>
                 <value>${kafka.consumer.topic.credit.for.lease}</value>
                 <value>${loan.application.feedback.topic}</value>
             </list>
         </constructor-arg>
         <property name="messageListener" ref="kafkaConsumerService" />
     </bean>
     <!-- 5.消費者並發消息監聽容器，執行doStart()方法 -->
     <bean id="messageListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart" >
         <constructor-arg ref="consumerFactory" />
         <constructor-arg ref="containerProperties" />
         <property name="concurrency" value="${concurrency}" />
     </bean>

1. consumerProperties-》consumerFactory 載入配置構造消費者工廠
2. messageListener-》containerProperties 載入容器配置（topics）
3. consumerFactory+containerProperties-》messageListenerContainer 容器配置（topics）+消息監聽器，構造一個並發消息監聽容器，並執行初始化方法doStart
4. 需要注意. KafkaConsumerSerivceImpl 此類 需要實現 MessageListener 接口

消費消息：

方案1：直接實現MessageListener接口，復寫onMessage方法，實現自定義消費業務邏輯。

 public class KafkaConsumerSerivceImpl implements MessageListener<String, String> {
     @Override
     public void onMessage(ConsumerRecord<String, String> data) {
         //根據不同主題，消費
         if("主題1".equals(data.topic())){
             //邏輯1
         }else if("主題2".equals(data.topic())){
             //邏輯2
         }
     }
 }

方案2：使用@KafkaListener注解，並設置topic,支持SPEL表達式。這樣方便拆分多個不同topic處理不同業務邏輯。（特別是有自己的事務的時候，尤其方便）

import org.springframework.kafka.annotation.KafkaListener;

public class KafkaConsumerSerivceImpl {
    @KafkaListener(topics = "${templar.aggrement.agreementWithhold.topic}")
    void templarAgreementNoticewithhold(ConsumerRecord<String, String> data){
       //消費業務邏輯
    }
}

 

Java注解方式

• 生產者

配置：

 /**
  * @description kafka 生產者配置
  */
 @Configuration
 @EnableKafka
 public class KafkaProducerConfig {
     public KafkaProducerConfig(){
         System.out.println("kafka生產者配置");
     }

     @Bean
     public ProducerFactory<Integer, String> producerFactory() {
         return new DefaultKafkaProducerFactory(producerProperties());
     }
 
     @Bean
     public Map<String, Object> producerProperties() {
         Map<String, Object> props = new HashMap<String, Object>();
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertiesUtil.getInstance().getString("kafka.producer.bootstrap.servers"));
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, PropertiesUtil.getInstance().getString("kafka.producer.key.serializer"));
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,PropertiesUtil.getInstance().getString("kafka.producer.value.serializer"));
         props.put(ProducerConfig.RETRIES_CONFIG,PropertiesUtil.getInstance().getInt("kafka.producer.retries"));
         props.put(ProducerConfig.BATCH_SIZE_CONFIG,PropertiesUtil.getInstance().getInt("kafka.producer.batch.size",1048576));
         props.put(ProducerConfig.LINGER_MS_CONFIG,PropertiesUtil.getInstance().getInt("kafka.producer.linger.ms"));
         props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,PropertiesUtil.getInstance().getLong("kafka.producer.buffer.memory",33554432L));
         props.put(ProducerConfig.ACKS_CONFIG,PropertiesUtil.getInstance().getString("kafka.producer.acks","all"));
         return props;
     }
 
     @Bean
     public KafkaTemplate<Integer, String> kafkaTemplate() {
         KafkaTemplate kafkaTemplate = new KafkaTemplate<Integer, String>(producerFactory(),true);
         kafkaTemplate.setDefaultTopic(PropertiesUtil.getInstance().getString("kafka.producer.defaultTopic","default"));
         return kafkaTemplate;
     }
 
 }

發送消息：

　　跟xml配置一樣。

• 消費者

配置：

 /**
  * @description kafka 消費者配置
  */
 @Configuration
 @EnableKafka
 public class KafkaConsumerConfig {
     public KafkaConsumerConfig(){
         System.out.println("kafka消費者配置加載...");
     }

     @Bean
     KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
     kafkaListenerContainerFactory() {
         ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                 new ConcurrentKafkaListenerContainerFactory();
         factory.setConsumerFactory(consumerFactory());
         factory.setConcurrency(3);
         factory.getContainerProperties().setPollTimeout(3000);
         return factory;
     }
 
     @Bean
     public ConsumerFactory<Integer, String> consumerFactory() {
         return new DefaultKafkaConsumerFactory(consumerProperties());
     }
 
     @Bean
     public Map<String, Object> consumerProperties() {
         Map<String, Object> props= new HashMap<String, Object>();
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertiesUtil.getInstance().getString("kafka.consumer.bootstrap.servers"));
         props.put(ConsumerConfig.GROUP_ID_CONFIG,  PropertiesUtil.getInstance().getString("kafka.consumer.group.id"));
         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,  PropertiesUtil.getInstance().getString("kafka.consumer.enable.auto.commit"));
         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, PropertiesUtil.getInstance().getString("kafka.consumer.auto.commit.interval.ms"));
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,  PropertiesUtil.getInstance().getString("kafka.consumer.session.timeout.ms"));
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,  PropertiesUtil.getInstance().getString("kafka.consumer.key.deserializer"));
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,  PropertiesUtil.getInstance().getString("kafka.consumer.value.deserializer"));
         return props;
     }
 
     @Bean
     public KafkaConsumerListener kafkaConsumerListener(){
         return new KafkaConsumerListener();
     }
 
 }

消費消息：

　　跟xml配置一樣。

 

 

引用：

https://www.cnblogs.com/hsc13-lxy14/p/9211224.html

https://docs.spring.io/spring-kafka/reference/htmlsingle/