如何配置频率Kafka消费者投票在sping-kafka

Question

我试图在我的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;
}

Answer 1

在2.3版本中，添加了一个名为idleBetweenPolls的新属性。

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

通过这种方式，您可以添加消费者调查之间的间隔。

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

Answer 2

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

要仅按需获取消息，您有两种选择：

• 使用消费者工厂创建消费者，订阅（或分配）主题/分区，并调用poll（）

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

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