Spring与Kafka整合

Posted 2021-11-23 投身大海的魚

一、导入依赖

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();
    }
  }
}