zoukankan      html  css  js  c++  java
  • Spring-Kafka —— 实现批量消费和手动提交offset

    spring-kafka的官方文档介绍,可以知道自1.1版本之后,

    @KafkaListener开始支持批量消费,只需要设置batchListener参数为true

    把application.yml中的enable-auto-commit设置为false,设置为不自动提交

     

    复制代码
    @Bean
    public KafkaListenerContainerFactory<?> batchFactory(ConsumerFactory consumerFactory){
        ConcurrentKafkaListenerContainerFactory<Integer,String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(10);
        factory.getContainerProperties().setPollTimeout(1500);
        factory.setBatchListener(true);//设置为批量消费,每个批次数量在Kafka配置参数中设置
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置手动提交ackMode
      return factory; 
    }
    复制代码
    复制代码
       //批量消息
        @KafkaListener(topics = {"first_top"},containerFactory="batchFactory")
        public void consumerBatch(List<ConsumerRecord<?, ?>> records, Acknowledgment ack){
            log.info("接收到消息数量:{}",record.size());
           //手动提交
          ack.acknowledge();
    }
    复制代码

    这里containerFactory = “batchFactory”要指定为批量消费

  • 相关阅读:
    Linux rpm 安装MySQL
    AOP 底层实现原理
    7 AOP
    HTTP 协议
    Oracle JDBC 标准连接实例
    Oracle JDBC 连接池
    Anaconda XGBoost安装
    Anaconda Spyder 导入自定义函数(模块)
    Hive常用sql
    决策树之信息增益计算模拟
  • 原文地址:https://www.cnblogs.com/suizhikuo/p/14011134.html
Copyright © 2011-2022 走看看