有没有办法配置@KafkaListener 的轮询间隔？答案

【问题标题】：Is there a way to configure polling interval of @KafkaListener?有没有办法配置@KafkaListener 的轮询间隔？
【发布时间】：2020-11-26 10:32:18
【问题描述】：

我是 Kafka 的 SpringBoot 领域的新手。我看到定义消费者的常用方法是@KafkaListener。
在 psat 中，我配置了一个调度程序，该调度程序每 X 次轮询 Kafka 代理，但在这里我没有找到如何指定此间隔或有关将数据从代理推送到消费者的文档。有没有办法配置它（或者如果使用数据的方式是通过推送我很乐意理解它）

【问题讨论】：

max.poll.interval.ms 是等待记录的最长时间...消费者在轮询请求之间没有这样的计时器，否则它被视为不活动，并重新平衡
@OneCricketeer 检查答案，最近的更改允许这种行为很好

标签： spring-boot apache-kafka spring-kafka

【解决方案1】：

在 spring kafka 中是 recently introduced，所以如果你使用的是现代版本，它应该是这样的：

build.gradle

implementation("org.springframework.kafka:spring-kafka:2.6.1") // or higher, minimum required is 2.3.0

Config.java

@Bean("singleKafkaListenerContainerFactoryManualCommit")
public ConcurrentKafkaListenerContainerFactory<?, ?> singleKafkaListenerContainerFactoryManualCommit(ConcurrentKafkaListenerContainerFactoryConfigurer configurer, ConsumerFactory<Object, Object> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.setBatchListener(false);
    factory.getContainerProperties().setIdleBetweenPolls(1000L); // <- this is basically it
    return factory;
}

它的样子：

2020-11-27 17:04:41.099  INFO 67146 --- [ntainer#0-0-C-1] c.r.t.m.d.l.service.KafkaService    : Consumed message 
2020-11-27 17:04:42.143  INFO 67146 --- [ntainer#0-0-C-1] c.r.t.m.d.l.service.KafkaService    : Consumed message
2020-11-27 17:04:43.185  INFO 67146 --- [ntainer#0-0-C-1] c.r.t.m.d.l.service.KafkaService    : Consumed message
2020-11-27 17:04:44.223  INFO 67146 --- [ntainer#0-0-C-1] c.r.t.m.d.l.service.KafkaService    : Consumed message

请记住，您需要 kafka 代理版本 >= 1.0 才能使其工作。

【讨论】：

idleBetweenPolls 在 2.3 中添加；它不依赖于代理版本，并且可以与所有最近的代理（> = 1.0）一起使用。 MANUAL_IMMEDIATEack模式也没有要求。