Spring Integration是否有开箱即用组件可批量分组RabbitMQ消息推送至Kinesis

有现成的组件可以直接实现该需求，具体使用方案如下：

• 批量分组可以直接使用 Spring Integration 核心提供的Aggregator组件，你只需要配置release-strategy为按消息数量触发即可实现按可配置的批次大小分组，同时可以额外配置group-timeout参数，避免消息量低峰期批次长时间无法出队的问题。
  如果你使用 Spring Integration AMQP 消费 RabbitMQ 消息，也可以直接开启消费侧批量消费配置，设置batch-size参数为你需要的批次大小，消费侧直接拉取指定数量的批量消息，不需要额外引入聚合器组件。
• 推送 Kinesis 侧可以直接使用 Spring Integration AWS 模块提供的KinesisMessageHandler，该组件原生支持接收PutRecordsRequestEntry集合作为输入，内部会自动调用AmazonKinesisAsync.putRecordsAsync方法完成批量提交，不需要自己手动封装批量调用逻辑。

以下是基于聚合器实现的参考配置：

// 消息聚合器配置，按数量+超时双条件触发批次释放
@Bean
public AggregatorFactoryBean messageBatchAggregator() {
    AggregatorFactoryBean aggregator = new AggregatorFactoryBean();
    // 所有消息归为同一个分组，不需要按维度拆分批次的场景可以直接用固定值
    aggregator.setCorrelationStrategy(message -> "GLOBAL_BATCH_GROUP");
    // 每累计100条消息释放一个批次
    aggregator.setReleaseStrategy(new MessageCountReleaseStrategy(100));
    // 最长等待时间2秒，超时后不满100条也释放
    aggregator.setGroupTimeoutExpression(new ValueExpression<>(2000L));
    aggregator.setSendPartialResultOnExpiry(true);
    return aggregator;
}

// Kinesis 批量发送处理器
@Bean
public MessageHandler kinesisBatchHandler(AmazonKinesisAsync amazonKinesisAsync) {
    KinesisMessageHandler handler = new KinesisMessageHandler(amazonKinesisAsync, "你的Kinesis流名称");
    handler.setAsync(true);
    return handler;
}

注意：如果使用聚合器做批量分组，并且RabbitMQ消费侧开启了手动ACK，需要在批次成功推送到Kinesis之后再统一ACK该批次对应的所有RabbitMQ消息，避免消息丢失。

内容的提问来源于stack exchange，提问作者LBPS