控制Kafka听众消费

Question

所以首先，为了能够暂停/停止消费者，我必须访问MessageListenerContainer。这意味着，在配置中，我将创建：ConcurrentKafkaListenerContainerFactory并（从2.2开始）使用它创建ConcurrentMessageListenerContainer的托管bean。然后可以使用这个bean来启动/停止消费者。管用。一旦它是并发的...我假设，我传递给setupMessageListener的必须是无状态类的实例，以便可以从多个线程/使用者对其进行操作。因此，如果我想注入spring依赖项，就像我以前在bean上使用@kafkalistener注释方法一样，我可以在这里传递无状态单例bean实例。

现在关于重新定位：这似乎很容易。只需在messageListener类中实现ConsumerSeekAware，该类是通过setupMessageListener注册的，并存储回调。然后，您可以只需autowire ConsumerSeekAware messageListener singleton，并执行搜索。相关的MessageListenerContainer，不管分区/并发设置如何，都应该执行查找。因为当ConcurrentMessageListenerContainer以大于1的并发开始时，它会在所有分区上启动多个KafkaMessageListenerContainer（？），但是由于所有分区都共享相同的groupId，所以只有一个使用者会使用消息，或者b）每个KafkaMessageListenerContainer将有一些分区子集。但是在seek中，我们必须指定Topic+Partition+offset，所以无论在哪种情况下，seek都应该由适当的KafkaMessageListenerContainer来拾取。

对吧？

我知道这是令人生畏的短信/问题，但也许这对其他人也有用。

Answer 1

您不需要创建自己的容器

@KafkaListener(id = "foo", topics = " ... "

然后，获取对KafkalistenerEndpointRegistry（自动连线等）的引用。

然后

registry.getListenerContainer("foo").stop();

registery.stop();