spring-integration-kafka集成配置2.x版本

「爱情、让人受尽委屈。」 2022-01-06 06:45 329阅读 0赞

# 前言 #

最近项目用到kafka，项目之前是用的spring-integration-kafka 1.x版本的集成，但是因为1.x版本使用的zookeeper为kafka消费者的连接地址,因为kafka版本升级，导致后面各种kafka消息的延迟和消息的丢失，然后就考虑升级spring-integration-kafka的版本到2.x版本，但是这本以为就是改一下maven pom文件version版本号的事，谁知道这一改就捅了马蜂窝！！！前前后后搞了差不多一个月，期间各种百度，翻墙google，github仓库搜索，官方文档翻查，都没有找到完整的集成配置消费者consumer的例子。期间有放弃过一次，不用spring-integration-kafka，改为直接用spring-kafka，改完测试、上线一切OK！本以为终于可以睡一个安稳觉的时候，各种kafka消息积压报警短信纷涌而至（每天几百万的消息量，处理不过来。。。）。

因为spring-kafka只支持多线程监听收取消息，且设置的线程数与kafka集群的分区数对应，如果线程数比监听数多了也只是浪费系统资源，当时我们的kafka集群是有8台机器，大概40个分区，因此只能配置40个listen监听，也就是40个消费线程，多配也没用。而之前用spring-integration-kafka是可以配置无限个消费线程的（只要你的机器能支撑住），如果坚持要使用spring-kafka的话需要配置配置一个线程池，监听线程拿到消息后，交给线程池的去执行，监听线程只负责拿到消息，具体的业务处理交给另一个线程处理（spring-integration-kafka已经帮我们这样封装了）。不得已啊，只能重新捡起spring-integration-kafka 2.x版本这个坑继续研究

**这里贴一下github的demo地址：**

[https://github.com/legend-c/spring-integration-kafka-demo.git][https_github.com_legend-c_spring-integration-kafka-demo.git]

觉得不错的就点进去右上角给个star 满足一下我的虚荣心吧，哈哈哈哈哈

下面就消费者的配置做一下演示吧，生产者的有空再整理一下上传到 github 去

### 一、项目结构 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NnaF9CYWJ5_size_16_color_FFFFFF_t_70][]

### 二、maven依赖 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NnaF9CYWJ5_size_16_color_FFFFFF_t_70 1][]

### 三、java consumer消费者类 ###

package com.cai;
    
    import org.springframework.kafka.support.Acknowledgment;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
    
    import lombok.extern.slf4j.Slf4j;
    @Slf4j
    public class MyConsumer {
    	public void onMessage(Message<String> message) {
    		MessageHeaders headers = message.getHeaders();
    		//手动提交偏移量的参数对象
    		Acknowledgment acknowledgment = (Acknowledgment) headers.get("kafka_acknowledgment");
    		
    		String topic = (String) headers.get("kafka_receivedTopic");
    		log.info("topic:{}",topic);
    		
    		long offset = (long) headers.get("kafka_offset");
    		log.info("offset:{}",offset);
    		
    		int partition = (int) headers.get("kafka_receivedPartitionId");
    		log.info("partition:{}",partition);
    		
    		String kafkaMsg = message.getPayload();
    		log.info("kafkaMsg:{}",kafkaMsg);
    		
    		//提交偏移量
    		acknowledgment.acknowledge();
    	}
    }

### 四、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:int="http://www.springframework.org/schema/integration"
        xmlns:task="http://www.springframework.org/schema/task" 
        xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
        xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka 
                            http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
                            http://www.springframework.org/schema/integration 
                            http://www.springframework.org/schema/integration/spring-integration.xsd
                            http://www.springframework.org/schema/beans 
                            http://www.springframework.org/schema/beans/spring-beans.xsd
                            http://www.springframework.org/schema/task  
                            http://www.springframework.org/schema/task/spring-task-4.1.xsd">
                           
        
        <bean id="myConsumer" class="com.cai.MyConsumer"/>                  
         
         
        <int:channel id="inputFromKafka">
        	<int:dispatcher task-executor="taskExecutor"/>
        </int:channel>
        
        
        <int:outbound-channel-adapter channel="inputFromKafka" ref="myConsumer" method="onMessage" />
             
        <task:executor id="taskExecutor" pool-size="50" keep-alive="200" rejection-policy="CALLER_RUNS" queue-capacity="100"/>
        
    	<int-kafka:message-driven-channel-adapter 
    		auto-startup="true" 
    		channel="inputFromKafka" 
    		listener-container="kafka-container" mode="record" />
    		
       
        <bean id="consumerProperties" class="java.util.HashMap">  
            <constructor-arg>  
                <map>  
                    <entry key="bootstrap.servers" value="${kafka-bootstrap.servers}"/>  
                    <entry key="group.id" value="${group-id}"/>  
                    <entry key="auto.offset.reset" value="${auto.offset.reset}"/>  
                    <entry key="enable.auto.commit" value="${enable.auto.commit}"/>  
                    <entry key="auto.commit.interval.ms" value="${auto.commit.interval.ms}"/>  
                    <entry key="session.timeout.ms" value="${session.timeout.ms}"/>  
                    <entry key="max.poll.records" value="${max.poll.records}"/>  
                    <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 bean="consumerProperties"/>  
            </constructor-arg>  
        </bean>  
      
        <bean id="kafka-container" class="org.springframework.kafka.listener.KafkaMessageListenerContainer" >  
            <constructor-arg name="consumerFactory" ref="consumerFactory"/>
            <constructor-arg name="containerProperties" ref="containerProperties"/>
        </bean>
        <bean id="containerProperties" class="org.springframework.kafka.listener.config.ContainerProperties">  
            <constructor-arg value="${kafka-topic}"/>
            
            <property name="ackMode" value="MANUAL_IMMEDIATE"/>
        </bean>
        
    </beans>

### 五、config配置文件 ###

kafka-topic=test-topic
    group-id=groud-v1
    
    #下面的参数配置都是spring-kafka的配置，百度就知道这些参数的意义了
    kafka-bootstrap.servers=192.168.56.101:9092,192.168.56.101:9093,192.168.56.101:9094
    auto.offset.reset=earliest
    enable.auto.commit=false
    auto.commit.interval.ms=1000
    session.timeout.ms=15000
    kafka-listener-concurrency-num=30
    max.poll.records=200

好了，消费者配置就是这么简单了，有不懂的或错误的可以给我留言，欢迎指出一起学习交流！！！！

**这里再贴一下github的demo地址：**

[https://github.com/legend-c/spring-integration-kafka-demo.git][https_github.com_legend-c_spring-integration-kafka-demo.git]

觉得不错的就点进去右上角给个star 满足一下我的虚荣心吧，哈哈哈哈哈

### 六、参考资料 ###

[https://blog.csdn.net/qq\_27808011/article/details/80108622][https_blog.csdn.net_qq_27808011_article_details_80108622]

https://docs.spring.io/spring-integration/docs/5.0.4.RELEASE/reference/html/

https://www.aliyun.com/jiaocheng/301276.html

https://blog.csdn.net/xiayutai1/article/details/53302652?locationNum=4&fps=1

http://www.iteye.com/topic/744524

https://blog.csdn.net/slivefox/article/details/3740541

https://my.oschina.net/zhzhenqin/blog/86586

http://www.importnew.com/16538.html

[https_github.com_legend-c_spring-integration-kafka-demo.git]: https://github.com/legend-c/spring-integration-kafka-demo.git
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NnaF9CYWJ5_size_16_color_FFFFFF_t_70]: /images/20211227/c5b6a2c8203643348b501d5589474c65.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NnaF9CYWJ5_size_16_color_FFFFFF_t_70 1]: /images/20211227/d49ae13d3c04469bb9254792e50dc086.png
[https_blog.csdn.net_qq_27808011_article_details_80108622]: https://blog.csdn.net/qq_27808011/article/details/80108622