kafka原理五之springboot 集成批量消费

这篇具有很好参考价值的文章主要介绍了kafka原理五之springboot 集成批量消费。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

前言

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

二、yaml配置文件

三、消息消费

手动提交非批量消费

  String 类型接入

使用注解方式获取消息头、消息体

手动提交批量消费

ConsumerRecord类接收

String类接收

使用注解方式获取消息头、消息体,则也是使用 List 来接收:

并发消费 

配置类方式

四、Kafka参数调优

一、Consumer参数说明

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

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



前言

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


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

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

二、yaml配置文件

spring:

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

三、消息消费

手动提交非批量消费

  •   String 类型接入
 @KafkaListener(topics = {"test-topic"}, groupId = "test-consumer-group")
    public void onMessage(String message, Consumer consumer) {
        System.out.println("接收到的消息:" + message);
        consumer.commitSync();
    }
  • 使用注解方式获取消息头、消息体
         /**
         * 处理消息
         */
        @KafkaListener(topics = "test-topic", groupId = "test-consumer-group")
        public void onMessage(@Payload String message,
                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                              @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                              @Header(name = KafkaHeaders.RECEIVED_MESSAGE_KEY, required = false) String key,
                              @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
                              Acknowledgment ack) {

            try {
                
                ack.acknowledge();
                log.info("Consumer>>>>>>>>>>>>>end");
            } catch (Exception e) {
                log.error("Consumer.onMessage#error . message={}", message, e);
                throw new BizException("事件消息消费失败", e);
            }
        } 

手动提交批量消费

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

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

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

  • ConsumerRecord类接收
    /**
     * kafka的批量消费监听器
     */
    @KafkaListener(topics = "test-topic", groupId = "test-consumer-group")
    public void onMessage(List<ConsumerRecord<String, String>> records, Consumer consumer) {
        try {
            log.info("Consumer.batch#size={}", records == null ? 0 : records.size());

            if (CollectionUtil.isEmpty(records)) {
                //分别是commitSync(同步提交)和commitAsync(异步提交)
                consumer.commitSync();
                return;
            }

            for (ConsumerRecord<String, String> record : records) {
                String message = record.value();

                if (StringUtils.isBlank(message)) {
                    continue;
                }

               //处理业务数据
               //doBuiness();
            }

            consumer.commitSync();
            log.info("Consumer>>>>>>>>>>>>>end");
        } catch (Exception e) {
            log.error("Consumer.onMessage#error .", e);
            throw new BizException("事件消息消费失败", e);
        }
    }
  • String类接收
 @KafkaListener(topics = {"test-topic"}, groupId = "test-consumer-group")
    public void onMessage(List<String> message, Consumer consumer) {
        System.out.println("接收到的消息:" + message);
        consumer.commitSync();
    }
  • 使用注解方式获取消息头、消息体,则也是使用 List 来接收:
@Component
public class KafkaConsumer {
    // 消费监听
    @KafkaListener(topics = {"test-topic"})
    public void listen2(@Payload List<String> data,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<String> keys,
                        @Header(KafkaHeaders.RECEIVED_TIMESTAMP) List<Long> tss) {
        System.out.println("收到"+ data.size() + "条消息:");
        System.out.println(data);
        System.out.println(topics);
        System.out.println(partitions);
        System.out.println(keys);
        System.out.println(tss);
    }
}

并发消费 

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

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

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

配置类方式

通过自定义配置类的方式也是可以的,但是相对yml配置来说还是有点麻烦的

/**
 * 消费者配置
 */
@Configuration
public class KafkaConsumerConfig {
 
    /**
     * 消费者配置
     * @return
     */
    public Map<String,Object> consumerConfigs(){
        Map<String,Object> props = new HashMap<>();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9002");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }
 
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        //并发数量
        factory.setConcurrency(3);
        //开启批量监听
        factory.setBatchListener(true);
        return factory;
    }
}

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

@KafkaListener(topics = {"test-topic"},containerFactory = "batchFactory")
public void consumer(List<String> message){
	System.out.println("接收到的消息:" + message);
}

四、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拉取新消息,这样可以做到按照消费能力拉取消息,减轻消费者的压力。文章来源地址https://www.toymoban.com/news/detail-628633.html

到了这里,关于kafka原理五之springboot 集成批量消费的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 分布式 - 消息队列Kafka:Kafka 消费者的消费位移

    01. Kafka 分区位移 对于Kafka中的分区而言,它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。偏移量从0开始,每个新消息的偏移量比前一个消息的偏移量大1。 每条消息在分区中的位置信息由一个叫位移(Offset)的数据来表征。分区位移总是从 0 开始,假设一

    2024年02月12日
    浏览(46)
  • 分布式 - 消息队列Kafka:Kafka 消费者消费位移的提交方式

    最简单的提交方式是让消费者自动提交偏移量,自动提交 offset 的相关参数: enable.auto.commit:是否开启自动提交 offset 功能,默认为 true; auto.commit.interval.ms:自动提交 offset 的时间间隔,默认为5秒; 如果 enable.auto.commit 被设置为true,那么每过5秒,消费者就会自动提交 poll() 返

    2024年02月12日
    浏览(43)
  • 分布式 - 消息队列Kafka:Kafka消费者和消费者组

    1. Kafka 消费者是什么? 消费者负责订阅Kafka中的主题,并且从订阅的主题上拉取消息。与其他一些消息中间件不同的是:在Kafka的消费理念中还有一层消费组的概念,每个消费者都有一个对应的消费组。当消息发布到主题后,只会被投递给订阅它的每个消费组中的一个消费者

    2024年02月13日
    浏览(41)
  • 分布式消息队列Kafka(四)- 消费者

    1.Kafka消费方式 2.Kafka消费者工作流程 (1)总体工作流程 (2)消费者组工作流程 3.消费者API (1)单个消费者消费 实现代码 (2)单个消费者指定分区消费 代码实现: (3)消费者组消费 复制上面CustomConsumer三个,同时去订阅统一个主题,消费数据,发现一个分区只能被一个

    2023年04月26日
    浏览(44)
  • 消息队列-Kafka-消费方如何分区与分区重平衡

    消费分区 资料来源于网络 消费者订阅的入口:KafkaConsumer#subscribe 消费者消费的入口:KafkaConsumer#poll 处理流程: 对元数据重平衡处理:KafkaConsumer#updateAssignmentMetadataIfNeeded 协调器的拉取处理:onsumerCoordinator#poll 执行已完成的【消费进度】提交请求的回调函数:invokeCompletedOff

    2024年04月14日
    浏览(32)
  • 分布式 - 消息队列Kafka:Kafka消费者的分区分配策略

    Kafka 消费者负载均衡策略? Kafka 消费者分区分配策略? 1. 环境准备 创建主题 test 有5个分区,准备 3 个消费者并进行消费,观察消费分配情况。然后再停止其中一个消费者,再次观察消费分配情况。 ① 创建主题 test,该主题有5个分区,2个副本: ② 创建3个消费者CustomConsu

    2024年02月13日
    浏览(42)
  • celery笔记五之消息队列的介绍

    本文首发于公众号:Hunter后端 原文链接:celery笔记五之消息队列的介绍 前面我们介绍过 task 的处理方式,将 task 发送到队列 queue,然后 worker 从 queue 中一个个的获取 task 进行处理。 task 的队列 queue 可以是多个,处理 task 的 worker 也可以是多个,worker 可以处理任意 queue 的 t

    2024年02月09日
    浏览(42)
  • Spring-Kafka如何实现批量消费消息并且不丢失数据

    先给答案: 某个业务对象由多张表关联而成,要创建该对象需要向多张表插入数据,基于canal的监控就会有多次该对象的变更记录,而Kafka消费的时候也会多次处理同一个对象(虽然不同表,但是同一个对象的不同部分),原有的Kafka消费者是一次处理一条,这将造成重复对同

    2024年02月13日
    浏览(46)
  • 分布式 - 消息队列Kafka:Kafka消费者分区再均衡(Rebalance)

    01. Kafka 消费者分区再均衡是什么? 消费者群组里的消费者共享主题分区的所有权。当一个新消费者加入群组时,它将开始读取一部分原本由其他消费者读取的消息。当一个消费者被关闭或发生崩溃时,它将离开群组,原本由它读取的分区将由群组里的其他消费者读取。 分区

    2024年02月12日
    浏览(37)
  • SpringBoot: RabbitMQ消息队列之同时消费多条消息

    1. basicQos预取方法参数解析 basicQos(int prefetchCount) basicQos(int prefetchCount, boolean global) basicQos(int prefetchSize, int prefetchCount, boolean global) 参数: prefetchSize:可接收消息的大小 prefetchCount:处理消息最大的数量。 global:是不是针对整个Connection的,因为一个Connection可以有多个Channel,如

    2024年02月01日
    浏览(73)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包