@KafkaListener搭配@Scheduled定时消费仅获单条，如何一次性拉取Topic全量消息

问题根因 #

你当前单次触发只能拉取1条消息是两个问题共同导致的：

• 配置文件中max.poll.records参数设置为1，Kafka消费者每次拉取最多只能拿到1条消息
• 消费方法consumeToSnapshot中每处理1条消息就直接调用listenerContainer.stop()停止监听器，剩余未消费的消息自然无法被拉取

解决方案 #

1. 调整Kafka消费者配置

修改application.yml中的max.poll.records参数，设置为你预期单次拉取的最大消息量，比如1000：

"[max.poll.records]": 1000

如果需要避免这个配置影响其他消费者，可以单独为当前@KafkaListener指定独立的消费配置。

2. 改造为批量消费模式

首先修改Kafka配置类，新增支持批量消费的监听容器工厂：

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchKafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true); // 开启批量监听
    return factory;
}

然后修改业务实现类的消费方法，处理完所有拉取到的消息后再停止监听器：

@Override
@KafkaListener(
    id = "snapshotOfOutagesId", 
    topics = Constants.KAFKA_TOPIC, 
    groupId = "snapshotOfOutages", 
    autoStartup = "false",
    containerFactory = "batchKafkaListenerContainerFactory" // 指定批量监听工厂
)
public void consumeToSnapshot(List<ConsumerRecord<String, OutageDTO>> records) {
    log.info("本次定时拉取到Kafka消息条数：{}", records.size());
    // 遍历处理所有拉取到的消息
    for (ConsumerRecord<String, OutageDTO> cr : records) {
        String content = cr.value().toString();
        JSONObject jsonObject= new JSONObject(content);
        Map<String, Object> outageMap = jsonToMap(jsonObject);
        brokerProducerService.sendMessage(globalConfig.getTopicProperties().getSnapshotTopicName(),
                outageMap.get("outageId").toString(), toJson(outageMap));
    }
    // 所有消息处理完成后再停止监听器
    MessageListenerContainer listenerContainer = registry.getListenerContainer("snapshotOfOutagesId");
    listenerContainer.stop();
}

补充说明 #

如果Topic中待消费的消息量可能超过你设置的max.poll.records，可以增加逻辑判断：如果本次拉取的消息条数等于max.poll.records，说明还有未拉取的消息，暂不停止监听器，直到拉取到的消息数小于max.poll.records再停止，即可保证单次触发拉取完全部待消费消息。
如果你的需求是每次定时任务都拉取Topic全量历史消息而非增量消息，可以在监听器启动时手动将offset seek到分区起始位置，并且关闭自动offset提交即可。

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