Spring@KafkaListener和并发

Question

我正在与spring boot spring@KafkaListener合作。我期望的行为是：我的Kafka侦听器以10个线程读取消息。因此，如果其中一个线程挂起，其他消息将继续读取和处理消息。

我定义了bean of

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory)
{

    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.getContainerProperties().setMissingTopicsFatal(false);
    factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.INFO);
    return factory;
}

和spring启动配置：

spring.kafka.listener.concurrency=10

我看到所有配置都能正常工作，我在jmx中看到了我的10个线程：

但是我做了这样的测试：

 @KafkaListener(topics = {
            "${topic.name}" }, clientIdPrefix = "${kafka.client.id.prefix}", idIsGroup = false, id = "${kafka.listener.name}", containerFactory = "kafkaListenerContainerFactory")
    public void listen(ConsumerRecord<String, String> record)
    {
        if(record.getVersion() < 3) {
            try {
                Thread.sleep(20000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        else
            System.out.println("It works!");

    }

如果版本是

也许我的期望不是真的，这是Kafka听众的正确行为。请帮我处理Kafka的并发性，为什么会这样？

Answer 1

如果你想拥有故障转移Kafka，你必须启动更多的应用程序实例。

示例：你有一个名为test的主题，有一个分区，你将用同一个Kafka组创建两个应用实例。一个实例将处理您的数据，另一个将等待并开始处理消息，以防第一个实例崩溃。如果您有N个分区，其中包含应用程序的N 1或2或3个实例，则情况相同。此外，每个实例将只有一个使用者线程。

欲了解更多信息，请在谷歌上搜索：Kafka消费群体。

Answer 2

kafka不是这样工作的；您至少需要和消费者一样多的分区（由Spring容器中的concurrency控制）。

而且，一次只能有一个消费者（在一个组中）从一个分区消费，因此，即使增加分区，其他消费者也不会收到“卡住”消费者后面同一分区中的记录。