Spring與Kafka整合

一、匯入依賴

Spring版本：4.2.6.RELEASE

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

二、配置生產者和消費者

2.1 生產者

<?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">


  <!-- 定義producer的引數 -->
  <bean id="producerProperties" class="java.util.HashMap">
    <constructor-arg>
      <map>
        <!-- bootstrap.servers的ip地址和埠號-->
        <entry key="bootstrap.servers" value="10.0.40.11:9092"/>
        <entry key="group.id" value="cms"/>
        <entry key="retries" value="3"/>
        <entry key="batch.size" value="16384"/>
        <!--
          當生產端積累的訊息達到batch-size或接收到訊息linger.ms后,生產者就會將訊息提交給kafka
          linger.ms為0表示每接收到一條訊息就提交給kafka,這時候batch-size其實就沒用了
          ‘1’表示每次批量一條
        -->
        <entry key="linger.ms" value="1"/>
        <entry key="buffer.memory" value="33554432"/>
        <entry key="acks" value="all"/>
        <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="producerProperties"/>
  </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="test"/>
  </bean>
</beans>

2.2 消費者

<?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">
  <!-- 1.定義consumer的引數 -->
  <!-- 基本的配置引數   可以寫成組態檔或者這種${bootstrap.servers} 組態檔獲取的  可以區分開發測驗環境    -->
  <bean id="consumerProperties" class="java.util.HashMap">
    <constructor-arg>
      <map>
        <entry key="bootstrap.servers" value="10.0.40.11:9092"/>
        <!-- 指定消費組名 -->
        <entry key="group.id" value="cms-test"/>
        <entry key="enable.auto.commit" value="false"/>
        <entry key="auto.offset.reset" value="earliest"/>
        <entry key="session.timeout.ms" value="15000"/>
        <entry key="request.timeout.ms" value="18000"/>
        <entry key="max.poll.records" value="1"/>
        <entry key="heartbeat.interval.ms" value="6000"/>
        <entry key="auto.commit.interval.ms" value="1000"/>
        <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>

  <!-- 創建工廠  然后把配置資訊注入-->
  <bean id="consumerFactory"
        class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
    <constructor-arg ref="consumerProperties"/>
  </bean>

  <!-- 把實際消費的類關聯進來 -->
  <bean id="kafkaConsumerListener" class="cn.com.test.common.kafka.KafkaConsumerListener"/>

  <!-- 然后把這個類和消費的topic注入這個container  topic也配置成靈活的 -->
  <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">
    <constructor-arg name="topics">
      <list>
        <!--消費的topic主題名稱-->
<!--        <value>nwqxt</value>-->
        <value>test</value>
      </list>
    </constructor-arg>
    <!--手動提交-->
    <property name="ackMode" value="MANUAL_IMMEDIATE"/>
    <property name="messageListener" ref="kafkaConsumerListener"/>
  </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="2" />
  </bean>-->
  <!-- 單執行緒訊息監聽容器 -->
  <bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer"
        init-method="doStart" lazy-init="true">
    <constructor-arg ref="consumerFactory"/>
    <constructor-arg ref="containerProperties"/>
  </bean>
</beans>

2.3 加入配置

在spring-context.xml檔案中加入配置

<import resource="spring-kafka-producer.xml"/>
<import resource="spring-kafka-consumer.xml"/>

三、消費者監聽器

/**
 * className: KafkaConsumerSerivceImpl <br>
 * description: 手動確認，回執
 *
 * @author guobc
 * @version 1.0.0
 * @date 2021-08-09 16:45
 */
@Slf4j
public class KafkaConsumerListener implements AcknowledgingMessageListener<String, String> {

  @Override
  public void onMessage(ConsumerRecord<String, String> data, Acknowledgment ack) {
    try {
      log.info("Kafka消費訊息為：{}", data);
      ack.acknowledge();
    } catch (Exception e) {
      log.error("Kafka訊息消費例外，錯誤原因為：{}", e.getMessage());
    } finally {
      ack.acknowledge();
    }
  }
}