Spring Boot中如何配置Kafka Consumer每2小时消费一次消息

你当前的配置无法实现「消费一条后暂停2小时」的需求，反而触发了频繁重平衡且不消费消息，核心问题出在配置逻辑的误解和参数冲突上，以下是具体修复方案：

问题根源 #

1. idleBetweenPolls使用错误：该参数仅在poll返回空消息（无数据可消费）时生效，有消息时会立即发起下一轮poll，无法实现消费后的强制暂停。
2. 参数冲突导致重平衡：你设置MAX_POLL_INTERVAL_MS=7200000（2小时），但如果消费逻辑+等待时间超过这个值，消费者会被集群判定为「死亡」，触发分区重平衡。
3. 未限制单次拉取消息数：默认情况下一次poll会拉取多条消息，无法实现逐条消费后暂停的逻辑。
4. 手动提交未落实：关闭了自动提交但未手动确认偏移量，导致偏移量不更新，重平衡后重复拉取或无法拉取新消息。

修复步骤 #

1. 调整消费者核心配置 #

添加单次拉取消息数限制，扩大最大轮询间隔以容纳暂停时间：

@Bean
public Map<String, Object> scnConsumerConfigs() {
    Map<String, Object> propsMap = new HashMap<>();

    // common props
    logger.info("KM Dataloader :: Kafka Brokers for Software topic: {}", bootstrapServersscn);
    propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersscn);
    propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    // 关闭自动提交后，该参数无意义，直接移除
    // propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
    // 限制每次poll仅拉取1条消息
    propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
    // 设置最大轮询间隔为2小时1分钟，避免暂停2小时触发超时
    propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 7260000);

    // ssl props
    propsMap.put("security.protocol", mpaasSecurityProtocol);
    propsMap.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);
    propsMap.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePassword);
    propsMap.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keystorePath);
    propsMap.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, keystorePassword);
    return propsMap;
}

2. 修改容器配置 #

移除无效的idleBetweenPolls，开启手动提交模式，并确保并发数不超过topic分区数：

ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
LOGGER.info("Setting concurrency to {} for {}", config.getConcurrency(), topicName);
// 并发数不能超过topic分区数，否则多余消费者会触发不必要的重平衡
factory.setConcurrency(Math.min(config.getConcurrency(), getTopicPartitionCount(SOFTWARE_TOPIC)));
factory.setConsumerFactory(cFactory);
factory.setRetryTemplate(retryTemplate);
// 移除无意义的idleBetweenPolls配置
// factory.getContainerProperties().setIdleBetweenPolls(7200000);
// 开启手动提交模式
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
return factory;

注：getTopicPartitionCount方法需要自行实现，用于获取指定topic的分区数量，若为单分区topic可直接设置为1。

3. 修改消费逻辑 #

添加手动提交偏移量，并在消费完成后强制暂停2小时：

@Bean
public KmKafkaListener softwareKafkaListener(KmSoftwareService softwareService) {
    return new KmKafkaListener(softwareService) {
        @KafkaListener(topics = SOFTWARE_TOPIC, containerFactory = "softwareMessageContainer", groupId = SOFTWARE_CONSUMER_GROUP)
        public void onscnMessageforSA20(@Payload ConsumerRecord<String, Object> record, Acknowledgment ack)
                throws InterruptedException {
            try {
                this.onMessage(record);
                // 手动提交偏移量，确保消费进度被记录
                ack.acknowledge();
            } finally {
                // 消费完成后暂停2小时
                Thread.sleep(7200000);
            }
        }
    };
}

关键注意事项 #

• 若使用重试模板，需确保重试逻辑不会重复触发暂停操作，可调整重试的触发条件或在重试回调中跳过暂停。
• Thread.sleep会阻塞当前消费者线程，若需要更优雅的暂停方式，可使用定时任务配合消费者暂停/恢复API，但上述方案已满足基础需求。
• 若topic存在多分区，每个分区的消费者会独立执行「消费-暂停」逻辑，需根据业务需求调整并发数和分区策略。

内容的提问来源于stack exchange，提问作者Kaustabh Nag