批量消费设置

@KafkaListener(
            id = "${spring.kafka.consumer.notice.id}",
            topicPartitions = {@TopicPartition(topic = "${spring.kafka.consumer.notice.topic}", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "0"))},
            groupId = "dvsfdvasdva",
            clientIdPrefix = "exactness-notice-batch",
            containerFactory = "batchFactory"
    )
    public void batchListener(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment) {
         for (ConsumerRecord<?, ?> record : records) {
             try {
                NoticeModel noticeModels = (NoticeModel)JsonUtil.jsonToBean((String) record.value(), NoticeModel.class);
                iBigdateService.exactnessMessageNoticeOne(noticeModels);
            } catch (Exception e) {
                log.error("编号:{},消费数据data:{},异常:{}",uuid,record.value(),e);
            }
        }
        acknowledgment.acknowledge();
    }



    @Bean
    @ConditionalOnMissingBean(name = "batchFactory")
    KafkaListenerContainerFactory<?> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new
                ConcurrentKafkaListenerContainerFactory<>();
        //并发数量
        factory.setConcurrency(concurrency);
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        // 开启批量监听
        factory.setBatchListener(batchListener);
        factory.getContainerProperties().setPollTimeout(pollTimeout);
        if (!enableAutoCommit) {//非自动提交需设置ACK方式
            factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        }
        //配置kafka监听是否自动启动
        factory.setAutoStartup(kafkaListenerFlag);
        return factory;
    }

`KafkaListener参数`	作用
id	消费者的id，当GroupId没有被配置的时候，默认id为GroupId
groupId	消费组ID，当前消费组，（小心修改导致重复消费）
`containerFactory`	配置的是监听容器工厂，也就是ConcurrentKafkaListenerContainerFactory，配置BeanName
`topics`	需要监听的Topic，可监听多个将topics 配置为：@KafkaListener(topics = “#{’${topics}’.split(’,’)}”)
`topicPartitions`	可配置更加详细的监听信息，必须监听某个Topic中的指定分区，或者从offset为200的偏移量开始监听 `topicPartitions = {@TopicPartition(topic = "${spring.kafka.consumer.notice.topic}", partitionOffsets = @PartitionOffset(partition = "1"//指定监听的分区, initialOffset = "0"//初始偏移量))},`
`topicPattern`	topicPattern 已经可以做到定期检查topic列表，然后将新加入的topic分配至某个消费者。 topicPattern = "test_topic2.*"
clientIdPrefix	消费者Id前缀
`containerGroup`
`errorHandler`	监听异常处理器，配置BeanName
`idIsGroup`	id是否为GroupId
`beanRef`	真实监听容器的BeanName，需要在 BeanName前加 "__"
`concurrency`	线程数量，消费者个数(注意，消费者数要小于等于你开的所有topic的分区数总和)
`autoStartup`	是否自动启动 true to auto start, false to not auto start.

关闭自动消费

Spring-Boot-kafka的配置中有一个参数，提供了自动消费的功能。使用自动消费我们在收到消息的时候会自动向kafka确认消费。但是在一些特殊场景中我们可能需要使用一些其他的消费模式。

enable-auto-commit: true

kafka的消费模式

需要自己配置AckMode时候的配置

spring:
  application:
    name: base.kafka
  kafka:
    bootstrap-servers: kafka服务地址1:端口,kafka服务地址2:端口,kafka服务地址3:端口
    producer:
      # 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，
      # 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
      retries: 0
      #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
      #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
      #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
      #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
      #可以设置的值为：all, -1, 0, 1
      acks: 1
    consumer:
      group-id: testGroup
      # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
      auto-offset-reset: earliest
      # 设置自动提交offset
      enable-auto-commit: false
      max-poll-records: 2
server:
  port: 8060

kafka支持的消费模式，设置在AbstractMessageListenerContainer.AckMode的枚举中，下面就介绍下各个模式的区别

AckMode模式	作用
MANUAL	当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
MANUAL_IMMEDIATE	手动调用Acknowledgment.acknowledge()后立即提交
RECORD	当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
BATCH	当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
TIME	当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
COUNT	当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
COUNT_TIME	TIME或COUNT 有一个条件满足时提交

MANUAL 和 MANUAL_IMMEDIATE

MANUAL

当每一条记录被消费者监听器（ListenerConsumer）处理之后提交

监听器工厂的配置

    /**
     * MANUAL   当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
     * @param consumerFactory
     * @return
     */
    @Bean("manualListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setPollTimeout(1500);
        factory.setBatchListener(true);
        //配置手动提交offset
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
        return factory;
    }

指定监听器使用的配置

    /**
     * MANUAL   当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
     * @param message
     * @param ack
     */
    @KafkaListener(containerFactory = "manualListenerContainerFactory" , topics = "kafka-manual")
    public void onMessageManual(List<Object> message, Acknowledgment ack){
        log.info("manualListenerContainerFactory 处理数据量：{}",message.size());
        message.forEach(item -> log.info("manualListenerContainerFactory 处理数据内容：{}",item));
        ack.acknowledge();//直接提交offset
    }

MANUAL_IMMEDIATE

当每一条记录被消费者监听器（ListenerConsumer）处理之后提交

监听器工厂的配置

    /**
     * MANUAL_IMMEDIATE 手动调用Acknowledgment.acknowledge()后立即提交
     * @param consumerFactory
     * @return
     */
    @Bean("manualImmediateListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualImmediateListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setPollTimeout(1500);
        factory.setBatchListener(true);
        //配置手动提交offset
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

指定监听器使用的配置

    /**
     * MANUAL_IMMEDIATE 手动调用Acknowledgment.acknowledge()后立即提交
     * @param message
     */
    @KafkaListener(containerFactory = "manualImmediateListenerContainerFactory" , topics = "kafka-manualImmediate")
    public void onMessageManualImmediate(List<Object> message, Acknowledgment ack){
        log.info("manualImmediateListenerContainerFactory 处理数据量：{}",message.size());
        message.forEach(item -> log.info("manualImmediateListenerContainerFactory 处理数据内容：{}",item));
        ack.acknowledge();//直接提交offset
    }

MANUAL和MANUAL_IMMEDIATE两者的相同和区别

相同之处

这两种模式都是需要进行手动确认ack.acknowledge();才能完成消息的消费，否则在重启消费端实例的时候数据会再次被消费端接收到。

两者的区别

MANUAL: 在处理完最后一次轮询的所有结果后，将队列排队，并在一次操作中提交偏移量。可以认为是在批处理结束时提交偏移量
MANUAL_IMMEDIATE：只要在侦听器线程上执行确认，就立即提交偏移。会在批量执行的时候逐一提交它们。

关于两者的解释可以看看这个提问 what-is-the-difference-between-manual-and-manual-immediate-in-spring-kafka-ackmo

RECORD

监听器工厂的配置

    /**
     * RECORD   当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
     * @param consumerFactory
     * @return
     */
    @Bean("recordListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> recordListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setPollTimeout(1500);
        //配置手动提交offset
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.RECORD);
        return factory;
    }

指定监听器使用的配置

    /**
     * RECORD   当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
     * @param message
     */
    @KafkaListener(containerFactory = "recordListenerContainerFactory" , topics = "kafka-record")
    public void onMessageRecord(List<Object> message){
        log.info("recordListenerContainerFactory 处理数据量：{}",message.size());
        message.forEach(item -> log.info("recordListenerContainerFactory 处理数据内容：{}",item));
    }

关于AckMode.RECORD 模式

使用RECORD模式的时候，当监听器处理完消息后会自动处理，使用此模式不需要手动消费。

TIME

监听器工厂的配置

    /**
     * TIME     当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
     * @param consumerFactory
     * @return
     */
    @Bean("timeListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> timeListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setPollTimeout(2000);
        //配置手动提交offset
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.TIME);
        return factory;
    }

指定监听器使用的配置

    /**
     * TIME     当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
     * @param message
     */
    @KafkaListener(containerFactory = "timeListenerContainerFactory" , topics = "kafka-time")
    public void onMessageTime(List<Object> message){
        log.info("timeListenerContainerFactory 处理数据量：{}",message.size());
        message.forEach(item -> log.info("timeListenerContainerFactory 处理数据内容：{}",item));
    }

关于AckMode.TIME 模式

此模式会在监听器监听到消息后的PollTimeout的时间段内提交消费请求。上面配置中将PollTimeout设置的超时时间为2秒。使用请求http://localhost:8060//ack/sendStr/kafka-time/3发送三条消息，在消费实例接收到消息后在PollTimeout到达之前关闭消费端实例，再次启动消费实例会发现之前消费的消息会被再次读取到。而接收到消息PollTimeout之后关闭消费实例，则会发现消息已经被成功消费。

COUNT

监听器工厂的配置

    /**
     * COUNT    当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
     * @param consumerFactory
     * @return
     */
    @Bean("countListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> countListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setPollTimeout(1500);
        factory.getContainerProperties().setAckCount(5);
        //配置手动提交offset
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.COUNT);
        return factory;
    }

指定监听器使用的配置

    /**
     * COUNT    当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
     * @param message
     */
    @KafkaListener(containerFactory = "countListenerContainerFactory" , topics = "kafka-count")
    public void onMessageCount(List<Object> message){
        log.info("countListenerContainerFactory 处理数据量：{}",message.size());
        message.forEach(item -> log.info("countListenerContainerFactory 处理数据内容：{}",item));
    }

关于AckMode.COUNT 模式

此模式会在监听器监听到消息等于AckCount的数量提交消费请求。上面配置中将`AckCount``设置为5条消息。使用请求http://localhost:8060//ack/sendStr/kafka-count/9向队列发送9条数据，然后关闭消费端实例，然后再次启动时发现最后4条消息再次被读取到。此时前5条数据被消费，最后四条消息需要凑齐5条数据后才能消费。

原文链接：https://blog.csdn.net/weixin_43870064/article/details/109066756