如何在spring-kafka中配置kafka消费者轮询的频率答案

【问题标题】：How to configure frequency for kafka consumer poll in spring-kafka如何在spring-kafka中配置kafka消费者轮询的频率
【发布时间】：2019-01-30 09:45:33
【问题描述】：

我正在尝试在我的 Spring Boot 项目中使用 spring-kafka 从我的 kafka 中读取消息。我正在使用@KafkaListener，但问题是我的消费者一直在运行。一旦我从控制台生成一条消息，它就会在我的应用程序中弹出。我想定期轮询。我怎样才能做到这一点？

@Service
public class KafkaReciever {

private static final Logger LOGGER =
        LoggerFactory.getLogger(KafkaReciever.class);

private CountDownLatch latch = new CountDownLatch(1);

public CountDownLatch getLatch() {
    return latch;
}

@KafkaListener(topics = "test")
public void receive(String payload) {
    LOGGER.info("received payload='{}'", payload);
    latch.countDown();
}

}

这是我的消费者配置：

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

    // list of host:port pairs used for establishing the initial connections to the Kafka cluster
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    // allows a pool of processes to divide the work of consuming and processing records
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "foo1");

    // automatically reset the offset to the earliest offset
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    return props;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());

    return factory;
}

【问题讨论】：

标签： java spring-boot spring-kafka

【解决方案1】：

这就是它的设计方式；它是一个消息驱动容器（并且与其他 Spring 消息传递技术抽象 - RabbitMQ、JMS 等一致）。

要仅按需获取消息，您有 2 个选择：

使用消费者工厂创建消费者，订阅（或分配）主题/分区并调用poll()
使用spring-integration-kafka的KafkaMessageSource并致电receive()

在这两种情况下，如果您使用的是 kafka 组管理，则需要注意max.poll.interval.ms 以避免重新平衡。

您可以使用 spring 集成入站通道适配器定期轮询消息源。

【讨论】：

使用本文获取消费者的完整信息learning.oreilly.com/library/view/kafka-the-definitive/…

【解决方案2】：

从 2.3 版本开始，添加了一个名为 idleBetweenPolls 的新属性。

从 2.3 版开始，ContainerProperties 提供了一个 idleBetweenPolls 选项，让侦听器容器中的主循环在 KafkaConsumer.poll() 调用之间休眠。从提供的选项和 max.poll.interval.ms 消费者配置与当前记录批处理时间之间的差异中选择一个实际的睡眠间隔作为最小值。

这样您可以在消费者投票之间添加间隔。

来源：https://docs.spring.io/spring-kafka/reference/html

【讨论】：