春季集成 AWS RabbitMQ Kinesis

Question

我们有一个用例，从 RabbitMQ 消费消息，对其进行转换并将记录推送到 Kinesis 流。 RabbitMQ 中的每条消息对应于 Kinesis 流中的一条记录。 AmazonKinesisAsync 支持 PutRecordsRequestEntry，其中可以将多个记录分组以进行单个 HTTP 调用。

我的问题是 Spring Integration 中是否有任何开箱即用的组件可以帮助在推送到 Kinesis 之前根据可配置的记录大小将来自 rabbitmq 的消息分组为一个批次？

Answer 1

查看聚合器 EI 模式及其在 Spring Integration 中的实现：https://docs.spring.io/spring-integration/docs/current/reference/html/message-routing.html#aggregator。因此，在将消息推送到 KinesisMessageHandler 之前，您有一个 AggregatingMessageHandler 用于按照配置的大小从 RabbitMQ 批量处理这些消息并发出 PutRecordsRequest 容器对象。