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

【问题标题】：How can I create spring integration flow read from kafka with queue limit process (parallel)如何使用队列限制过程创建从kafka读取的spring集成流（并行）
【发布时间】：2020-04-09 21:36:08
【问题描述】：

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

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

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

块引用

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

【问题讨论】：

标签： spring-boot spring-integration spring-kafka spring-integration-dsl

【解决方案1】：

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

相反，要提高并发性，请增加主题中的分区数，并设置侦听器容器并发性以获得您需要的消费者数量（例如 50 个）。

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

【讨论】：

您好，我需要为许多租户处理库存更新（有 1,000 名客户同时运行），如果我也增加分区数量，此更新有一个需要一定时间的外部服务调用很多，kafka打开很多文件描述符，
想象一下，每个卖家每个周期都有 100 个带有库存更新的产品，如果有 1000 个卖家要进行可接受的竞争，就必须有一个包含许多分区的主题，并查看 Confluence 的建议许多分区需要对 kafka 进行非常大的监控，有 10 个分区我没有足够的并行度来获得性能。有了这个，他想象不阻止消息的接收并通过队列进行有限数量的并行
不要将代码/配置放在 cmets 中；太难读了。改为编辑问题。对来自单个分区的记录执行异步处理是不切实际的。
设置并发后（您的配置看起来正确），您应该看到 n 个线程 ...#0-0-C-1、...#0-1-C-1 - 您引用 0-1-C-1 的事实意味着您至少有两个线程。如果您必须对每个消费者进行异步操作，那么 Kafka 可能不是您的正确选择。您打算如何在这样的环境中可靠地管理消费者偏移量？