Spring 卡夫卡 : Poll for new messages instead of being notified using `onMessage`

Question

我在我的项目中使用 Spring Kafka，因为在基于 Spring 的项目中使用 Kafka 消息似乎是一个自然的选择。要使用消息，我可以使用 MessageListener 接口(interface)。 Spring Kafka 在内部负责为每条新消息调用我的 onMessage 方法。

但是，在我的设置中，我更喜欢显式轮询新消息并按顺序处理它们(这将需要几秒钟)。作为解决方法，我可能只是在我的 onMessage 实现中阻塞，或者在内部缓冲消息。然而，这似乎违背了Spring Kafka的核心思想。

Kafka 的设计使得消费者必须轮询新消息，这符合我的要求。有没有办法利用 Spring Kafka 的这种“自然”工作流程？

我应该避免在这个用例中使用 Spring Kafka 吗？

KafkaConsumer documentation状态:

For use cases where message processing time varies unpredictably, neither of these options may be sufficient. The recommended way to handle these cases is to move message processing to another thread, which allows the consumer to continue calling poll while the processor is still working. Some care must be taken to ensure that committed offsets do not get ahead of the actual position. Typically, you must disable automatic commits and manually commit processed offsets for records only after the thread has finished handling them (depending on the delivery semantics you need). Note also that you will need to pause the partition so that no new records are received from poll until after thread has finished handling those previously returned.

相关问题:https://github.com/spring-projects/spring-kafka/issues/195

Answer 1

必须不断轮询消费者的问题现已解决(在 0.10.1.x 中由 KIP-62 解决)，因此这不再是问题(只要您不超过 max. poll.interval.ms)，默认为 5 分钟，但可以增加。

但是，如果您想自行轮询，您仍然可以使用 spring-kafka(例如，如果您使用 Boot，则获得 Spring Boot 自动配置的优点)，但您可以从以下位置获取 Consumer直接使用 DefaultKafkaConsumerFactory 和 poll() 。