是否有可能有一个Kafka消费者线程每个主题？

Question

ConsumerThread1-[topic1-0,topic2-0,topic3-0]
ConsumerThread2-[topic1-1,topic2-1,topic3-1]

但是，我们希望每个主题有一个使用者线程，而不是每个分区有一个KafkaListener（或使用者线程）。例如：

ConsumerThread1-[topic1-0,topic1-1]
ConsumerThread2-[topic2-0,topic2-1]
ConsumerThread3-[topic3-0,topic3-1]

有什么办法可以做到这一点吗？

Answer 1

您可以从Spring-Kafka:2.2中为每个主题创建单独的容器，并将并发性设置为1，以便每个容器将从每个主题中消耗

从版本2.2开始，您可以使用同一个工厂创建任何ConcurrentMessageListenerContainer。如果您希望创建多个具有相似属性的容器，或者希望使用一些外部配置的工厂，例如Spring Boot自动配置提供的工厂，这可能会很有用。创建容器后，可以进一步修改其属性，其中许多属性是通过使用container.getContainerProperties（）设置的。下面的示例配置ConcurrentMessageListenerContainer：

@Bean
html" target="_blank">public ConcurrentMessageListenerContainer<String, String>(
    ConcurrentKafkaListenerContainerFactory<String, String> factory) {

ConcurrentMessageListenerContainer<String, String> container =
    factory.createContainer("topic1", "topic2");
container.setMessageListener(m -> { ... } );
return container;
}

注意：以这种方式创建的容器不会添加到endpoint注册表中。应该将它们创建为@bean定义，以便在应用程序上下文中注册它们。