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

Question

需要帮助吗

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

我有以下架构

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

一个包含5个EC2实例的Amazon堆栈，每个实例都有tomcat服务器，运行一个Spring Boot应用程序，该应用程序具有Spring集成流，该集成流使用Spring集成aws的SQS消息驱动通道适配器来消费SQS的消息(https://github.com/spring-projects/spring-integration-aws)

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

SQS队列-

限制Amazon SQS允许客户端批量读取最多10条消息的请求，但我可以有多个客户端并行使用更多的消息。

在Amazon SQS中，需要手动删除消息。这是使用spring integration完成的，我只在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" />

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

我试着把




Answer 1

为了达到这样的要求，您可以使用executor频道，而不是默认的DirectChannel。

这样，所有的SQS消息都将被分发到执行器通道提供的线程中，因此并行执行。

有关执行器频道的更多信息，请参阅参考手册。

使现代化

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

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

使现代化

如果你仍然坚持要几个SQS适配器，那么简化版本是这样的：

<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，并开始使用其<代码> ITNGRESATORACTION上下文<代码> >动态<代码>集成流< /代码>注册：https://docs.spring.io/spring-integration/docs/5.0.4.RELEASE/reference/html/java-dsl.html#java-dsl运行时流