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

kafka原理之springboot 集成批量消费_springboot kafka 多线程消费

作者：IT小白 | 2024-04-14 02:52:39

赞

踩

springboot kafka 多线程消费

前言

由于 Kafka 的写性能非常高，因此项目经常会碰到 Kafka 消息队列拥堵的情况。遇到这种情况，我们可以通过并发消费、批量消费的方法进行解决。

一、新建一个maven工程，添加kafka依赖

  1. <dependency>
  2. 		<groupId>org.springframework.kafka</groupId>
  3. 		<artifactId>spring-kafka</artifactId>
  4. </dependency>

二、yaml配置文件

  1. spring:
  2.  
  3.   kafka:
  4.     bootstrap-servers: 127.0.0.1:9002
  5.     producer:
  6.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  7.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  8.     consumer:
  9.       group-id: test-consumer-group
  10.       # 当 Broker 端没有 offset（如第一次消费或 offset 超过7天过期）时如何初始化 offset，当收到 OFFSET_OUT_OF_RANGE 错误时，如何重置 Offset
  11.       # earliest：表示自动重置到 partition 的最小 offset
  12.       # latest：默认为 latest，表示自动重置到 partition 的最大 offset
  13.       # none：不自动进行 offset 重置，抛
  14.       auto-offset-reset: latest
  15.       # 是否在消费消息后将 offset 同步到 Broker，当 Consumer 失败后就能从 Broker 获取最新的 offset
  16.       enable-auto-commit: false
  17.       ## 当 auto.commit.enable=true 时，自动提交 Offset 的时间间隔，建议设置至少1000
  18.       auto-commit-interval: 2000
  19.       max-poll-records: 30
  20.       heartbeat-interval: 3000
  21.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  22.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  23.       properties:
  24.         # 使用 Kafka 消费分组机制时，消费者超时时间。当 Broker 在该时间内没有收到消费者的心跳时，认为该消费者故障失败，Broker 发起重新 Rebalance 过程。目前该值的配置必须在 Broker 配置group.min.session.timeout.ms=6000和group.max.session.timeout.ms=300000 之间
  25.         session.timeout.ms: 60000
  26.         # 使用 Kafka 消费分组机制时，消费者发送心跳的间隔。这个值必须小于 session.timeout.ms，一般小于它的三分之一
  27.         heartbeat.interval.ms: 3000
  28.         # 使用 Kafka 消费分组机制时，再次调用 poll 允许的最大间隔。如果在该时间内没有再次调用 poll，则认为该消费者已经失败，Broker 会重新发起 Rebalance 把分配给它的 partition 分配给其他消费者
  29.         max.poll.interval.ms: 300000
  30.         request.timeout.ms: 600000
  31.     listener:
  32.       # 在侦听器容器中运行的线程数。
  33.       concurrency: 2
  34.       type: batch
  35.       max-poll-records: 50
  36.       #当 auto.commit.enable 设置为false时，表示kafak的offset由customer手动维护，
  37.       #spring-kafka提供了通过ackMode的值表示不同的手动提交方式
  38.       #手动调用Acknowledgment.acknowledge()后立即提交
  39.       ack-mode: manual_immediate
  40.       # 消费者监听的topic不存在时，项目会报错，设置为false
  41.       missing-topics-fatal: false

三、消息消费

手动提交非批量消费

  •   String 类型接入
    1.  @KafkaListener(topics = {"test-topic"}, groupId = "test-consumer-group")
    2.     public void onMessage(String message, Consumer consumer) {
    3.         System.out.println("接收到的消息：" + message);
    4.         consumer.commitSync();
    5.     }

  • 使用注解方式获取消息头、消息体
  1.          /**
  2.          * 处理消息
  3.          */
  4.         @KafkaListener(topics = "test-topic", groupId = "test-consumer-group")
  5.         public void onMessage(@Payload String message,
  6.                               @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
  7.                               @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
  8.                               @Header(name = KafkaHeaders.RECEIVED_MESSAGE_KEY, required = false) String key,
  9.                               @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
  10.                               Acknowledgment ack) {
  11.  
  12.             try {
  13.                 
  14.                 ack.acknowledge();
  15.                 log.info("Consumer>>>>>>>>>>>>>end");
  16.             } catch (Exception e) {
  17.                 log.error("Consumer.onMessage#error . message={}", message, e);
  18.                 throw new BizException("事件消息消费失败", e);
  19.             }
  20.         }

 

手动提交批量消费

想要批量消费，首先要开启批量消费，通过listener.type属性设置为batch即可开启，看下代码吧：

  1. spring:
  2.   kafka:
  3.     consumer:
  4.       group-id: test-consumer-group
  5.       bootstrap-servers: 127.0.0.1:9092
  6.       max-poll-records: 50 # 一次 poll 最多返回的记录数
  7.     listener:
  8.       type: batch # 开启批量消费

 

如上设置了启用批量消费和批量消费每次最多消费记录数。这里设置 max-poll-records是50，并不是说如果没有达到50条消息，我们就一直等待。而是说一次poll最多返回的记录数为50

  • ConsumerRecord类接收
    1.     /**
    2.      * kafka的批量消费监听器
    3.      */
    4.     @KafkaListener(topics = "test-topic", groupId = "test-consumer-group")
    5.     public void onMessage(List<ConsumerRecord<String, String>> records, Consumer consumer) {
    6.         try {
    7.             log.info("Consumer.batch#size={}", records == null ? 0 : records.size());
    8.  
    9.             if (CollectionUtil.isEmpty(records)) {
    10.                 //分别是commitSync（同步提交）和commitAsync（异步提交）
    11.                 consumer.commitSync();
    12.                 return;
    13.             }
    14.  
    15.             for (ConsumerRecord<String, String> record : records) {
    16.                 String message = record.value();
    17.  
    18.                 if (StringUtils.isBlank(message)) {
    19.                     continue;
    20.                 }
    21.  
    22.                //处理业务数据
    23.                //doBuiness();
    24.             }
    25.  
    26.             consumer.commitSync();
    27.             log.info("Consumer>>>>>>>>>>>>>end");
    28.         } catch (Exception e) {
    29.             log.error("Consumer.onMessage#error .", e);
    30.             throw new BizException("事件消息消费失败", e);
    31.         }
    32.     }

  • String类接收
    1.  @KafkaListener(topics = {"test-topic"}, groupId = "test-consumer-group")
    2.     public void onMessage(List<String> message, Consumer consumer) {
    3.         System.out.println("接收到的消息：" + message);
    4.         consumer.commitSync();
    5.     }

  • 使用注解方式获取消息头、消息体，则也是使用 List 来接收：
    1. @Component
    2. public class KafkaConsumer {
    3.     // 消费监听
    4.     @KafkaListener(topics = {"test-topic"})
    5.     public void listen2(@Payload List<String> data,
    6.                         @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
    7.                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
    8.                         @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<String> keys,
    9.                         @Header(KafkaHeaders.RECEIVED_TIMESTAMP) List<Long> tss) {
    10.         System.out.println("收到"+ data.size() + "条消息：");
    11.         System.out.println(data);
    12.         System.out.println(topics);
    13.         System.out.println(partitions);
    14.         System.out.println(keys);
    15.         System.out.println(tss);
    16.     }
    17. }

  • 并发消费 

    再来看下并发消费，为了加快消费，我们可以提高并发数，比如下面配置我们将并发设置为 3。注意：并发量根据实际分区数决定，必须小于等于分区数，否则会有线程一直处于空闲状态

  1. spring:
  2.   kafka:
  3.     consumer:
  4.       group-id: test-consumer-group
  5.       bootstrap-servers: 127.0.0.1:9092
  6.       max-poll-records: 50 # 一次 poll 最多返回的记录数
  7.     listener:
  8.       type: batch # 开启批量监听
  9.       concurrency: 3 # 设置并发数

 

我们设置concurrency为3，也就是将会启动3条线程进行监听，而要监听的topic有5个partition，意味着将有2条线程都是分配到2个partition，还有1条线程分配到1个partition

配置类方式

通过自定义配置类的方式也是可以的，但是相对yml配置来说还是有点麻烦的(不提倡)

  1. /**
  2.  * 消费者配置
  3.  */
  4. @Configuration
  5. public class KafkaConsumerConfig {
  6.  
  7.     /**
  8.      * 消费者配置
  9.      * @return
  10.      */
  11.     public Map<String,Object> consumerConfigs(){
  12.         Map<String,Object> props = new HashMap<>();
  13.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
  14.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9002");
  15.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
  16.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  17.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  18.         return props;
  19.     }
  20.  
  21.     @Bean
  22.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> batchFactory() {
  23.         ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
  24.         factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
  25.         //并发数量
  26.         factory.setConcurrency(3);
  27.         //开启批量监听
  28.         factory.setBatchListener(true);
  29.         return factory;
  30.     }
  31. }

同时监听器通过@KafkaListener注解的containerFactory 配置指定批量消费的工厂即可,如下：

  1. 同时监听器通过@KafkaListener注解的containerFactory 配置指定批量消费的工厂即可,如下：
  2.

四、Kafka参数调优

一、Consumer参数说明


1、enable.auto.commit

该属性指定了消费者是否自动提交偏移量，默认值是true。
为了尽量避免出现重复数据（假如，某个消费者poll消息后，应用正在处理消息，在3秒后kafka进行了重平衡，那么由于没有更新位移导致重平衡后这部分消息重复消费）和数据丢失，可以把它设为 false，由自己控制何时提交偏移量。
如果把它设为true，还可以通过配置 auto.commit.interval.ms 属性来控制提交的频率。
 

2、auto.commit.interval.ms
自动提交间隔。范围：[0,Integer.MAX]，默认值是 5000 （5 s）

3、手动提交：commitSync/commitAsync
手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。

相同点：都会将本次poll的一批数据最大的偏移量提交。
不同点：commitSync会阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而commitAsync则没有失败重试机制，故有可能提交失败，导致重复消费。

4、max.poll.records
Consumer每次调用poll()时取到的records的最大数。


二、Kafka消息积压、消费能力不足怎么解决？

如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，同时相应的增加消费者实例，消费者数=分区数（二者缺一不可）。
如果是下游的数据处理不及时，则可以提高每批次拉取的数量，通过max.poll.records这个参数可以调整。
单个消费者实例的消费能力提升，可以用多线程/线程池的方式并发消费提高单机的消费能力。


三、Kafka消费者如何进行流控？

将自动提交改成手动提交（enable.auto.commit=false），每次消费完再手动异步提交offset，之后消费者再去Broker拉取新消息，这样可以做到按照消费能力拉取消息，减轻消费者的压力。
 

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

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

  

闽ICP备14008679号