如何在多个线程中执行 Spring 集成流以并行使用更多 Amazon SQS 队列消息？

Question

需要帮助

我需要创建多个并行执行的 sqs 队列消费者，但我不知道如何使用 Sprint 集成来实现这一点

我有以下架构

一个包含 20 万条消息的 Amazon SQS 队列

一个带有 5 个 EC2 实例的 Amazon 堆栈，每个实例都有一个 tomcat 服务器运行一个带有 Spring 集成流的 Spring Boot 应用程序，该流使用来自 spring-integration-aws 的 sqs-message-driven-channel-adapter (@ 987654321@)

并将消息发布到平均响应时间为 1 秒的 REST 服务（我无法修改 REST 服务是一个约束，但我可以并行发送消息）

SQS 队列 -> 堆栈（5 个 tomcat 实例） -> 休息服务

约束 Amazon SQS 允许客户端按请求批量读取消息，最大数量为 10 条消息，但我可以让多个客户端并行使用更多消息。

在 Amazon SQS 中，消息需要手动删除，这是使用 spring 集成完成的，我仅在 REST 服务返回 OK 时删除消息。

我对可能的重复没有问题（SQS 将相同的消息发送到两个不同的客户端）

我无法在我的 Spring Boot 应用程序中以任何方式存储消息

我的 Spring 集成流程

<aws-messaging:sqs-async-client id="clientWithCredentials"/>
<int-aws:sqs-message-driven-channel-adapter
  sqs="clientWithCredentials" 
  channel="channel_1"
  queues="https://sqs.us-east-1.amazonaws.com/123456789000/SomeAmazonSQSName"
  max-number-of-messages="10"/>

<int:channel id="channel_1" />
<int:outbound-channel-adapter ref="restService" method="publish" channel="channel_1" />

如何在多个线程中并行执行此流程以并行消耗更多消息？

我尝试将<int:poller fixed-rate="1" task-executor="executor" /> 放入 sqs-message-driven-channel-adapter 但不允许。

Answer 1

要实现这样的要求，您可以使用ExecutorChannel 而不是默认的DirectChannel。

这样，所有 SQS 消息都将分发到由ExecutorChannel 提供的线程，因此并行执行。

有关ExecutorChannel 的更多信息在Reference Manual 中。

更新

所以，我的建议应该反映在您当前的配置中，例如：

<int:channel id="channel_1">
   <int:dispatcher task-executor="someExecutor"/>
</int:channel>

更新

如果你还坚持有几个SQS Adapter，那么简化版是这样的：

<int-aws:sqs-message-driven-channel-adapter
    sqs="sqsAsyncClient" 
    channel="sqs-to-metricator"
    queues="https://sqs.us-east-1.amazonaws.com/123/SomeSQSQueueName"
    max-number-of-messages="10"
    />


<int-aws:sqs-message-driven-channel-adapter
    sqs="sqsAsyncClient" 
    channel="sqs-to-metricator"
    queues="https://sqs.us-east-1.amazonaws.com/123/SomeSQSQueueName"
    max-number-of-messages="10"
    />

<int-aws:sqs-message-driven-channel-adapter
    sqs="sqsAsyncClient" 
    channel="sqs-to-metricator"
    queues="https://sqs.us-east-1.amazonaws.com/123/SomeSQSQueueName"
    max-number-of-messages="10"
    />

<int:channel id="sqs-to-metricator" />

<int:outbound-channel-adapter ref="restService"
    method="publish" channel="sqs-to-metricator" />

为了避免重复，您可以考虑切换到 Java DSL 并开始使用其 ItengrationFlowContext 进行动态 IntegrationFlow 注册：https://docs.spring.io/spring-integration/docs/5.0.4.RELEASE/reference/html/java-dsl.html#java-dsl-runtime-flows