如何使用 Spring Integration 处理来自 AWS SQS FiFo 队列的 10 条以上并发消息答案

【问题标题】：How to process more than 10 concurrent messages from an AWS SQS FiFo queue using Spring Integration如何使用 Spring Integration 处理来自 AWS SQS FiFo 队列的 10 条以上并发消息
【发布时间】：2018-09-17 23:12:14
【问题描述】：

我希望能够使用 Spring 集成工作流一次处理 10 条以上的 SQS 消息。

根据这个问题，建议使用 ExecutorChannel。我更新了我的代码，但仍然有相同的症状。

How execute Spring integration flow in multiple threads to consume more Amazon SQS queue messages in parallel?

进行此更新后，我的应用程序请求 10 条消息并处理这些消息，只有在我在流程接近尾声时调用 amazonSQSClient.deleteMessage 后，它才会接受来自 SQS 的另外 10 条消息排队。

应用程序使用 SQS FiFo 队列。

还有什么我遗漏的，或者这是使用 SqsMessageDeletionPolicy.NEVER 然后在流程结束时删除消息的不可避免的症状？由于其他限制，在流程开始时接受消息并不是一个真正的选择。

这里是相关的sn-ps代码，做了一些简化，但我希望它能表达问题。

队列配置

@Bean
public AsyncTaskExecutor inputChannelTaskExecutor() {
    SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
    executor.setConcurrencyLimit(50);
    return executor;
}

@Bean
@Qualifier("inputChannel")
public ExecutorChannel inputChannel() {
    return new ExecutorChannel(inputChannelTaskExecutor());
}

我还尝试了 ThreadPoolTaskExecutor 而不是 SimpleAsyncTaskExecutor，结果相同，但我也会将其包括在内，以防它提供其他见解。

    @Bean
    public AsyncTaskExecutor inputChannelTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("spring-async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.afterPropertiesSet();
        executor.initialize();
        return executor;
    }

SQS 通道适配器

@Bean
public SqsMessageDrivenChannelAdapter changeQueueMessageAdapter() {
    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSQSClient, changeQueue);
    adapter.setOutputChannel(inputChannel);
    adapter.setMessageDeletionPolicy(SqsMessageDeletionPolicy.NEVER);
    return adapter;
}


@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerSpec poller() {
    return Pollers.fixedRate(500, TimeUnit.MILLISECONDS).maxMessagesPerPoll(10);
}

简化的主要流程

对我们来说，一个常见的场景是在短时间内获得大量的分支编辑。这个流程只“关心”至少发生了一次编辑。 messageTransformer 从有效负载文档中提取一个 id 并将其放在标题 dsp_docId 中，然后我们使用它来聚合（我们在其他几个地方使用这个 id，所以我们感觉标题是有意义的，而不是在自定义聚合器中完成所有工作）。

provisioningServiceActivator 检索分支的最新版本，然后路由器决定是否需要进一步转换（在这种情况下，它会将其发送到 transformBranchChannel）或者它可以发送到我们的 PI 实例（通过 sendToPiChannel）。

转换流程（未显示，我认为您不需要它）最终导致发送到 PI 流程，它只是先完成更多工作。

listingGroupProcessor 捕获所有 aws_receiptHandle 标头并将它们作为 |分隔列表。

sendToPi 流（和 errorFlow）以调用自定义处理程序结束，该处理程序负责删除该 aws_receiptHandle 字符串列表引用的所有 SQS 消息。

@Bean
IntegrationFlow sqsListener() {
    return IntegrationFlows.from(inputChannel)
                           .transform(messageTransformer)
                           .aggregate(a -> a.correlationExpression("1")
                                            .outputProcessor(listingGroupProcessor)
                                            .autoStartup(true)
                                            .correlationStrategy(message -> message.getHeaders().get("dsp_docId"))
                                            .groupTimeout(messageAggregateTimeout)  // currently 25s
                                            .expireGroupsUponCompletion(true)
                                            .sendPartialResultOnExpiry(true)
                                            .get())

                           .handle(provisioningServiceActivator, "handleStandard")
                           .route(Branch.class, branch -> (branch.isSuppressed() == null || !branch.isSuppressed()),
                                  routerSpec -> routerSpec.channelMapping(true, "transformBranchChannel")
                                                          .resolutionRequired(false)
                                                          .defaultOutputToParentFlow())

                           .channel(sendtoPiChannel)
                           .get();
}

【问题讨论】：

你能用同样的设置制作更多的SqsMessageDrivenChannelAdapter并发送到同一个频道吗？我的意思是让我们尝试遵循这个建议：stackoverflow.com/questions/46377333/…！
是的。使用上面定义的三个 SqsMessageDrivenChannelAdapter 实例（但在 Bean 名称的末尾有 1、2 和 3）它仍然一次只能读取 10 条消息。我对此感到特别困惑，因为我认为如果有点不雅的话，它会起作用。
那么，即使使用 3 个并发适配器，您仍然可以从 AWS SQS 获得最多 10 条消息？那么在我们确认（删除）那些拉取的 10 条消息之前，这确实是他们的限制......
开始看起来那样。令人沮丧。感谢您的时间。嗯嗯，周末好好想想B计划。

标签： java spring-integration amazon-sqs spring-integration-dsl spring-cloud-aws

【解决方案1】：

我想我会发布这个作为答案，因为这解决了我的问题，并且可能对其他人有所帮助。作为答案，它更有可能被发现，而不是对可能被忽略的原始问题进行编辑。

首先，我应该注意到我们正在使用 FiFo 队列。

问题实际上是更进一步的链，我们将 MessageGroupId 设置为描述数据源的简单值。这意味着我们有非常大的消息组。

从ReceiveMessage 文档中您可以看到，在这种情况下，它非常明智地阻止了您从该组请求更多消息，因为如果需要将消息放回队列中，则无法保证顺序。

更新发布消息的代码以设置适当的 MessageGroupId 然后意味着 ExecutorChannel 可以按预期工作。

虽然具有特定 MessageGroupId 的消息不可见，但在可见性超时到期之前，不会返回更多属于同一 MessageGroupId 的消息。只要另一个 MessageGroupId 也是可见的，您仍然可以接收具有另一个 MessageGroupId 的消息。

【讨论】：