spring-kafka KafkaListener中的并行处理和自动缩放答案

【问题标题】：Parallel processing and auto scaling in spring-kafka KafkaListenerspring-kafka KafkaListener中的并行处理和自动缩放
【发布时间】：2020-09-10 21:55:08
【问题描述】：

我正在使用 spring-kafka 来使用来自两个 Kafka 主题的消息，它们发送的消息格式如下。

    @KafkaListener(topics = {"topic_country1", "topic_country2"}, groupId = KafkaUtils.MESSAGE_GROUP)
    public void onCustomerMessage(String message, Acknowledgment ack) throws Exception {
        log.info("Message : {}  is received", message);
        ack.acknowledge();
    }

KafkaListener能否根据自己监听的topic数量和两个topic中的并行处理消息来分配消费者线程的数量？或者它不支持并行处理，消息必须在主题中等待，直到处理完一条消息？
如果主题中的消息数量较多，我需要自动扩展我的微服务以启动新实例（直到分区数量）。从 KafkaListener 的角度来看，我可以依靠哪些参数（CPU、内存）来找出主题中的消息数量更高？（即在 API 中，我可以通过监控 HTTP 延迟来自动扩展服务）

【问题讨论】：

标签： spring-boot apache-kafka kafka-consumer-api spring-kafka autoscaling

【解决方案1】：

可以设置concurrency属性运行更多线程；但每个分区只能由一个线程处理。要增加并发性，您必须增加每个主题中的分区数。在同一个监听器中监听多个主题时，如果这些主题只有一个分区，则可能无法获得您想要的并发，除非您更改 kafka 消费者分区分配器。

见https://docs.spring.io/spring-kafka/docs/2.5.0.RELEASE/reference/html/#using-ConcurrentMessageListenerContainer

在监听多个主题时，默认的分区分布可能不是你所期望的。例如，如果您有 3 个主题，每个主题有 5 个分区，并且您希望使用 concurrency=15，那么您只会看到 5 个活动消费者，每个消费者从每个主题分配一个分区，而其他 10 个消费者处于空闲状态。这是因为默认的 Kafka PartitionAssignor 是 RangeAssignor（参见它的 Javadoc）。对于这种情况，您可能需要考虑改用 RoundRobinAssignor，它将分区分配给所有消费者。然后，为每个消费者分配一个主题或分区。 ...

【讨论】：

如果我错了，请纠正我。如果我将并发级别设置为 15 并运行两个实例，第二个实例的侦听器会空闲吗？创建新线程以分配给分区我的要求是分配给主题。即如果@KafkaListener 监听两个分区，则应该有两个线程并行消耗。可能吗？还有哪些参数可以用来衡量自动缩放的高消息量？
我创建了 2 个主题，每个主题有 6 个分区，并使用并发 = 2 的同一个监听器来监听主题。在这种情况下，即使监听器创建了两个线程，线程的分配在分区中也是随机的两组。不会根据主题分配线程。
您可以实现自己的PartitionAssignor，也可以简单地在方法中添加两个@KafkaListener注解，每个注解一个主题；那么你会得到你想要的分布。