如何使用队列限制过程（并行）创建从kafka读取的spring集成流

Question

我想创建一个异步读取kafka消息的流，并使用队列通道累积大量要处理的消息，并且只有在处理完这些消息（例如50条消息）后，它才能处理另外50条消息，或者在释放队列中的空间时。我尝试使用一个从kafka委托读取到另一个流的流，该流具有一个带有PollerMetadata（Pollers.fixedDelay（500））的QueueChannel。maxMessagesPerPoll（50））但是轮询器使用一个线程来读取那里的消息，我无法并行处理这50条消息，如果我在轮询器中放置一个执行器，它将像正常的执行器一样工作，它将累积消息，并且不会挂起到50，直到我有一个新线程可供他从Kafka获取另一条消息。

目标是并行处理多达50条kafka消息，但他只在这个队列释放时在kafka（consumer.pool）中再次读取，但他无限地从kafka读取并在执行器或轮询器的限制量内进行处理，我可以使用带有kafka的Spring集成流来实现这个目标吗？

每个消费者主题只有这个配置就足够了？日志总是打印相同的线程：[ntainer#0-1-C-1]即使我为并发设置了10

块引用

> Kafka.messageDrivenChannelAdapter(consumerFactory,
> topic).configureListenerContainer { kafkaMessageListenerContainer ->  
> kafkaMessageListenerContainer.concurrency(concurrency)               
> kafkaMessageListenerContainer.ackMode(ContainerProperties.AckMode.RECORD)
> }
>                 .errorChannel(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)

Answer 1

您不应该使用队列通道或使用Kafka执行任何异步处理。跟踪主题/分区内的偏移量太难了。您将面临丢失消息的风险。

相反，要增加并发性，请增加主题中的分区数，并设置侦听器容器并发性以获得所需的使用者数（例如50）。

您通常应该拥有比消费者更多的分区，但您至少需要同样多的分区，因为组中只有一个消费者可以使用分区。