devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka多线程并发消费，批量消费_kafka spring pull多线程消费

作者：IT小白 | 2024-04-14 02:48:36

赞

踩

kafka spring pull多线程消费

注：上篇博客说了怎么搭建kafka集群，并实现springboot集成kafka实现消息单条消费，但是如果有场景需要批量消费，那么就需要对消费者进行配置了，并可以实现并发消费，即多个KafkaListener同时消费消息，以此达到批量消费和提高速度的目的。

1.上消费者配置

  1. @Configuration
  2. @EnableKafka
  3. public class KafkaConsumerConfig {
  4.  
  5.     @Value("${spring.kafka.bootstrap-servers}")
  6.     private String service;
  7.  
  8.     @Value("${spring.kafka.consumer.group-id}")
  9.     private String groupid;
  10.  
  11.     @Value("${spring.kafka.consumer.enable-auto-commit}")
  12.     private String autoCommit;
  13.  
  14.     @Value("${spring.kafka.consumer.auto-commit-interval}")
  15.     private String interval;
  16.  
  17.     // 默认发送心跳时间为10000ms，超时时间需要大于发送心跳时间
  18.     @Value("10000")
  19.     private String timeout;
  20.  
  21.     @Value("${spring.kafka.consumer.key-deserializer}")
  22.     private String keyDeserializer;
  23.  
  24.     @Value("${spring.kafka.consumer.value-deserializer}")
  25.     private String valueDeserializer;
  26.  
  27.     @Value("${spring.kafka.consumer.auto-offset-reset}")
  28.     private String offsetReset;
  29.  
  30.  
  31.     /**
  32.      * 获取kafka配置
  33.      * @return 配置map
  34.      */
  35.     private Map<String,Object> consumerConfig(){
  36.         Map<String,Object> props = new HashMap<>();
  37.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,service);
  38.         props.put(ConsumerConfig.GROUP_ID_CONFIG, groupid);
  39.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
  40.         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, interval);
  41.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,timeout);
  42.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
  43.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
  44.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,offsetReset);
  45.         return props;
  46.     }
  47.  
  48.     /**
  49.      * 获取工厂
  50.      * @return kafka工厂
  51.      */
  52.     private ConsumerFactory<String,String> consumerFactory(){
  53.         Map<String, Object> props = consumerConfig();
  54.         // 日志过滤入库一批量为1500条消息
  55.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,1500);// 单次poll的数量,批量消费时配置
  56.         return new DefaultKafkaConsumerFactory<>(consumerConfig());
  57.     }
  58.  
  59.     /**
  60.      * 实时推送使用的消费者工厂
  61.      * @return kafka消费者工厂
  62.      */
  63.     private ConsumerFactory<String,String> infoPushConsumerFactory(){
  64.         Map<String, Object> props = consumerConfig();
  65.         // 实时推送单次批量拉取数据设置为150
  66.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,150);
  67.         return new DefaultKafkaConsumerFactory<>(props);
  68.     }
  69.  
  70.     /**
  71.      * 获取kafka实例,该方法为单条消费
  72.      * @return kafka实例
  73.      */
  74.     @Bean(name = "kafkaListenerContainerFactory1")
  75.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>> kafkaListenerContainerFactory1(){
  76.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  77.         factory.setConsumerFactory(consumerFactory());
  78.         factory.setConcurrency(1);  // 连接池中消费者数量
  79.         factory.getContainerProperties().setPollTimeout(4000); //拉取topic的超时时间
  80.         factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);// 关闭ack自动提交偏移
  81.         System.out.println("调用的是自定义消费者池");
  82.         return factory;
  83.     }
  84.  
  85.     /**
  86.      * 获取kafka实例，该实例为批量消费
  87.      * @return kafka实例
  88.      */
  89.     @Bean(name = "kafkaListenerContainerFactory2")
  90.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>> kafkaListenerContainerFactory2(){
  91.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  92.         factory.setConsumerFactory(consumerFactory());
  93.         factory.setConcurrency(2);  // 连接池中消费者数量
  94.         factory.setBatchListener(true); // 是否并发消费
  95.         factory.getContainerProperties().setPollTimeout(4000); //拉取topic的超时时间
  96.         factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);// 关闭ack自动提交偏移
  97.         return factory;
  98.     }
  99.  
  100.     /**
  101.      * 实时推送获取的kafka实例，该实例为批量消费
  102.      * @return  kafka实例
  103.      */
  104.     @Bean(name = "infoPushKafkaListenerContainerFactory")
  105.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>> infoPushKafkaListenerContainerFactory(){
  106.         ConcurrentKafkaListenerContainerFactory<String,String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  107.         factory.setConsumerFactory(infoPushConsumerFactory());
  108.         factory.setConcurrency(10); // 连接池中消费者数量
  109.         factory.setBatchListener(true); // 是否并发消费
  110.         factory.getContainerProperties().setPollTimeout(4000);  // 拉取topic的超时时间
  111.         factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH); // 每次poll之前提交一次偏移
  112. //        factory.getConsumerFactory().getConfigurationProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,50); // 单次poll的数量
  113.         return factory;
  114.     }
  115. }

上述配置中，有多个消费者实例，在使用时指定bean名称即可

factory.setBatchListener(true)这行代码开启并发消费

factory.setConcurrency(2)这行代码声明有几个KafkaListener同时监听，达到多线程目的

如何手动管理偏移量？

先将自动提交设置为false

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);//这里的autoCommit为变量，我设置的false

然后设置消费工厂的提交模式

  1. factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);// 关闭ack自动提交偏移
  2.  
  3.  
  4.  
  5. //
  6. AckMode 如下:
  7.  
  8. RECORD :当listener一读到消息，就提交offset
  9.  
  10. BATCH : poll() 函数读取到的所有消息,就提交offset
  11.  
  12. TIME : 当超过设置的ackTime ，即提交Offset
  13.  
  14. COUNT ：当超过设置的COUNT，即提交Offset
  15.  
  16. COUNT_TIME ：TIME和COUNT两个条件都满足，提交offset
  17.  
  18. MANUAL ： Acknowledgment.acknowledge()即提交Offset，和Batch类似
  19.  
  20. MANUAL_IMMEDIATE： Acknowledgment.acknowledge()被调用即提交Offset

设置完后，可以再listen中进行回调，手动提交偏移量

  1. //批量消费
  2.     @KafkaListener(topics = {"topicone"},containerFactory="kafkaListenerContainerFactory2")
  3.     public void batchConsumer(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
  4.         System.out.println("此线程消费"+records.size()+"条消息----线程名:"+Thread.currentThread().getName());
  5.         records.forEach(record -> System.out.println("topic名称:"+record.topic()+"\n"+"分区位置:"+record.partition()+"\n"+"key:"+record.key()+"\n"+"偏移量:"+record.offset()+"\n"+"消息内容:"+record.value()));
  6.         ack.acknowledge();
  7.     }

下面看结果：

  1. topicone 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=topicone, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=key2, value=[{"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"core","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"encrypt","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"front","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}], timestamp=null), recordMetadata=topicone-1@45]
  2. topicone 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=topicone, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=key3, value=[{"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"core","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"encrypt","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"front","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}], timestamp=null), recordMetadata=topicone-1@46]
  3. topicone 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=topicone, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=key4, value=[{"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"core","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"encrypt","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"front","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}], timestamp=null), recordMetadata=topicone-1@47]
  4. topicone 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=topicone, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=key5, value=[{"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"core","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"encrypt","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"front","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}], timestamp=null), recordMetadata=topicone-1@48]
  5. topicone 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=topicone, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=key0, value=[{"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"core","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"encrypt","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"front","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}], timestamp=null), recordMetadata=topicone-0@37]
  6. topicone 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=topicone, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=key1, value=[{"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"core","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"encrypt","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}, {"appId":null,"userId":null,"sourceIp":"null","sourceUsername":"null","sourcePassword":"null","sourcePort":"null","sourcePath":"null","maxTps":null,"maxMemory":null,"appName":"front","localhostIp":"null","deployPath":"null","approvalStatus":null,"createTime":null,"updateTime":null,"resType":null,"projectType":"null","configurePath":"null","proxyIp":"null","proxyPath":"null","proxyUsr":"null","proxyPsw":"null","businessLine":"null"}], timestamp=null), recordMetadata=topicone-0@38]
  7. 此线程消费4条消息----线程名:org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1
  8. 此线程消费2条消息----线程名:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/419797
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号