使用 Spring Boot 为每个主题的 kafka 消费者设置轮询率

Question

是否可以在 SpringBoot 上为每个 kafka 主题的消费者设置不同的投票率？我希望有一个主题以更长的间隔（如 5 分钟）进行轮询。我希望用它来实现重试处理来自kafka的消息失败。

一个示例实现会有很大帮助。

Answer 1

从 2.3 版开始，监听器容器有了一个新属性 idleBetweenPolls。容器将确保轮询之间的时间不大于max.poll.interval.ms - 5000。

您可以使用容器定制器 bean 为特定消费者设置它...

@Component
class Customizer {

    public Customizer(ConcurrentKafkaListenerContainerFactory<?, ?> factory, DefaultKafkaProducerFactory<?, ?> pf) {
        factory.setContainerCustomizer(container -> {
            if (container.getContainerProperties().getGroupId().equals("slowGroup")) {
                // or you can use the topic(s)
                container.getContainerProperties().setIdleBetweenPolls(60_000);
            }
        });
    }

}