使用 SQS 和 Spring Boot 的消息聚合答案

【问题标题】：Message Aggregation using SQS and SpringBoot使用 SQS 和 Spring Boot 的消息聚合
【发布时间】：2021-06-01 16:58:03
【问题描述】：

我有一个用例/情况，其中 SQS（标准）将被消息淹没（500k+ 以北），微服务（基于弹簧启动）监听这些事件，使用它，并进行休息 API 调用（批处理-基于）到第 3 方 SaaS 系统（已附上相同的高级图）
这里的限制是 Spring Boot 消费者最多可以从 SQS 接收 10 条消息，转换有效负载，并使用这 10 条消息（记录）进行其余的 API 调用。

在进行其余 API 调用之前，有没有办法将这些消息聚合为 100 条消息（假设目标 SaaS 系统接受 100 条数据记录）？在这种情况下，春季批次会有所帮助吗？

我是否应该针对这种需求查看不同的堆栈？非常感谢任何帮助/指导。

谢谢

【问题讨论】：

Any help/guidance is much appreciated.。我添加了一个答案，它有帮助吗？如果是，请考虑接受：stackoverflow.com/help/someone-answers。否则，让我知道缺少什么来接受它。谢谢。

标签： spring-boot spring-batch amazon-sqs

【解决方案1】：

您所描述的实际上是 Spring Batch 的面向块的处理模型：可以从队列中读取项目，以 100 个项目的块（即可配置的块大小）累积并批量发布到您的 REST API模式。

Spring Batch 为您处理项目（以及更多）的分块。所以是的，即使我有偏见，我相信 Spring Batch 对您的用例来说是一个非常好的选择。

【讨论】：

嗨@Mahmoud Ben Hassine 做了一个基本的峰值检查使用弹簧批处理（块处理模型）从 SQS 读取的消息聚合，它正在工作。这是我上传到github的代码
@user9142486 很高兴听到它对您有用。感谢您分享代码！

【解决方案2】：

也许你应该试试 Spring Aggregator(Spring Integration)。

聚合器通过关联来组合一组相关消息并存储它们直到该组被认为是完整的。那时点，聚合器通过处理整个消息来创建单个消息分组并将聚合消息作为输出发送。

https://docs.spring.io/spring-integration/reference/html/aggregator.html

有关与 AWS 服务的春季集成，请参阅此 GitHub 存储库 https://github.com/spring-projects/spring-integration-aws/tree/main/src/test/java/org/springframework/integration/aws

我假设您有多个应用程序实例，并且可以在需要时轻松扩展（因为您有 50 万多条消息）。但是，您的应用程序仍然容易丢失数据。因此，构建一个可靠的系统总是具有挑战性的。由于您已经在云上，也许您应该考虑使用不同的云服务。

我认为对于您的情况，您应该看看 AWS Kinesis dataStream 和 Kinesis data fire软管。你可以参考这个， https://aws.amazon.com/blogs/big-data/stream-data-to-an-http-endpoint-with-amazon-kinesis-data-firehose/

【讨论】：