能否结合Poller与KafkaMessageDrivenChannelAdapter定时读取Kafka消息？

首先明确：KafkaMessageDrivenChannelAdapter 是基于推模式的组件，依赖 KafkaMessageListenerContainer 持续监听 Kafka 主题，一旦有消息就会被主动推送过来，它本身不支持 Poller 轮询机制，所以你配置的 PollerMetadata 不会生效。

为什么你的配置无效？ #

KafkaMessageListenerContainer 的工作逻辑是持续保持与 Kafka 集群的连接，实时监听主题的消息偏移量，有新消息就立即触发消费流程。Poller 是为拉模式组件（比如 KafkaMessageSource）设计的，用于控制主动拉取数据的频率，对推模式的监听容器完全不起作用。

正确实现定时拉取Kafka消息的方式 #

如果需要每隔 X 分钟从 Kafka 主动拉取一次消息，应该使用**拉模式的 KafkaMessageSource**搭配 Poller，示例代码如下：

1. 配置 KafkaMessageSource

@Bean
public KafkaMessageSource<String, String> kafkaMessageSource(ConsumerFactory<String, String> consumerFactory) {
    ConsumerProperties properties = new ConsumerProperties("your-target-topic");
    // 可选：配置批量拉取，一次获取多条消息
    properties.setMaxPollRecords(50);
    KafkaMessageSource<String, String> messageSource = new KafkaMessageSource<>(consumerFactory, properties);
    // 开启批量模式，让消息 payload 是 List 类型
    messageSource.setBatchMode(true);
    return messageSource;
}

2. 配置集成流与 Poller

@Bean
public IntegrationFlow kafkaPollingIntegrationFlow(KafkaMessageSource<String, String> kafkaMessageSource) {
    return IntegrationFlow.from(kafkaMessageSource, spec -> spec
                    .poller(poller -> poller.trigger(new PeriodicTrigger(5, TimeUnit.MINUTES))))
            .handle(message -> {
                // 这里编写你的消息处理逻辑
                List<String> messages = (List<String>) message.getPayload();
                System.out.println("定时拉取到 " + messages.size() + " 条消息");
            })
            .get();
}

替代方案：推模式下定时处理消息 #

如果因为业务限制必须使用 KafkaMessageDrivenChannelAdapter，但想实现类似“定时批量处理”的效果，可以通过下游队列通道 + Poller的方式：让监听容器实时把消息存入队列，再用 Poller 每隔 X 分钟从队列中取出消息处理。示例代码：

1. 配置队列通道和消息驱动适配器

@Bean
public QueueChannel bufferedMessageQueue() {
    return new QueueChannel();
}

@Bean
public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenAdapter(
        KafkaMessageListenerContainer<String, String> listenerContainer,
        QueueChannel bufferedMessageQueue) {
    KafkaMessageDrivenChannelAdapter<String, String> adapter = new KafkaMessageDrivenChannelAdapter<>(listenerContainer);
    adapter.setOutputChannel(bufferedMessageQueue);
    return adapter;
}

2. 配置 Poller 定时处理队列消息

@Bean
public IntegrationFlow queuePollingFlow(QueueChannel bufferedMessageQueue) {
    return IntegrationFlow.from(bufferedMessageQueue, spec -> spec
                    .poller(poller -> poller.trigger(new PeriodicTrigger(5, TimeUnit.MINUTES))))
            .handle(message -> {
                // 定时处理队列中的消息
                System.out.println("处理消息：" + message.getPayload());
            })
            .get();
}

注意：这种方案是实时接收消息并缓存，定时处理缓存的消息，和“定时从 Kafka 拉取消息”是不同的业务逻辑，需要根据实际需求选择。

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