spring-boot - Spring Boot 服务中具有长任务的 KafkaListener

Question

我有一个消费kafka消息并触发一个长进程的服务。处理一条消息可能需要 10 分钟以上。当前在“doSomething()”方法完成之前不会消耗新消息。我怎样才能使它成为并发处理并并行处理消息？

@Service
public class MyService {

    @KafkaListener(topics = "request-topic", groupId = "group_id"),
        containerFactory = "requirementsKafkaListenerFactory")    
    private void consumeKafkaRequirementsDataJson(KafkaRequirementsData kafkaRequirementsData) {
        System.out.println("Consumed JSON Message from kafka Topic: " + kafkaRequirementsData);
        doSomething(kafkaRequirementsData);
    }

Answer 1

您可以为此设置并发属性，但每个消费者的线程数量受限于该主题的分区数量。如果您已经有相当数量的分区，这可能是可行的方法，但我不会仅仅为了长任务的并行性而增加分区的数量。

您可以尝试在您的应用程序上实现多线程，但管理资源将由您负责，此外还要考虑到一旦消息被使用并且线程被启动，您将负责其管理(包括重试)如果出现问题)。

参见 https://howtoprogram.xyz/2016/05/29/create-multi-threaded-apache-kafka-consumer/和 https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/了解更多信息

TL;DR，我会根据以下因素做出决定:

1. 如果有足够多的分区可以并行运行而不会阻塞您的任务，请设置并发属性。
2. 如果没有足够的分区，将您的任务排入队列并使用线程池运行它们。