时间戳

首页 > 解决方案 > 如何在spring-kafka中配置kafka消费者轮询的频率

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

问题描述

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

标签： javaspring-bootspring-kafka

解决方案

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

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

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

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

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

推荐阅读