Spring Integration：如何将拆分后的行聚合为指定大小的批次？

嗨，要把转换后的JSON记录聚合成每25条一批发送到Kinesis，用Spring Integration的**聚合器（Aggregator）**就可以完美搞定！下面给你具体的配置示例和细节说明，完全适配你现有的CSV拆分、转JSON的流程：

核心实现思路 #

1. 你已经完成了CSV拆分、行转JSON的步骤，假设转换后的JSON消息输出到jsonRecordsChannel通道
2. 用聚合器组件收集这些JSON消息，直到攒够25条就自动释放批次
3. 额外配置超时机制，避免因为消息不足25条而一直等待（比如超过10秒就发送现有批次）

XML配置示例（延续你的现有配置） #

在你已有的配置基础上，添加聚合器和Kinesis发送组件即可：

<!-- 假设你已完成行转JSON，输出到这个通道 -->
<int:channel id="jsonRecordsChannel"/>

<!-- 聚合器：收集25条为一批，10秒超时兜底 -->
<int:aggregator id="batchAggregator"
                input-channel="jsonRecordsChannel"
                output-channel="kinesisBatchChannel"
                release-strategy-expression="size() == 25"
                group-timeout="10000"
                send-partial-result-on-expiry="true"/>

<!-- Kinesis 批量发送适配器 -->
<int-aws:kinesis-outbound-channel-adapter id="kinesisAdapter"
                                          channel="kinesisBatchChannel"
                                          kinesis-client="kinesisClient"
                                          stream="your-target-kinesis-stream"/>

关键配置参数说明： #

• release-strategy-expression="size() == 25"：当聚合的消息数量达到25条时，自动释放这个批次
• group-timeout="10000"：设置10秒超时，避免因消息不足25条导致积压，超时后发送当前已收集的所有消息
• send-partial-result-on-expiry="true"：开启超时后发送部分批次的功能，配合超时配置生效

Java注解配置示例（如果用注解方式） #

如果你的项目采用Java注解配置，也可以这样实现：

@Configuration
@EnableIntegration
public class KinesisBatchConfig {

    @Autowired
    private AmazonKinesis kinesisClient;

    // 转换后的JSON消息通道
    @Bean
    public MessageChannel jsonRecordsChannel() {
        return new DirectChannel();
    }

    // 聚合后的批量消息通道
    @Bean
    public MessageChannel kinesisBatchChannel() {
        return new DirectChannel();
    }

    // 聚合器配置
    @Bean
    public AggregatorFactoryBean batchAggregator() {
        AggregatorFactoryBean aggregator = new AggregatorFactoryBean();
        aggregator.setInputChannel(jsonRecordsChannel());
        aggregator.setOutputChannel(kinesisBatchChannel());
        // 按消息数量释放：攒够25条发送
        aggregator.setReleaseStrategy(new MessageCountReleaseStrategy(25));
        // 10秒超时兜底
        aggregator.setGroupTimeout(10000L);
        aggregator.setSendPartialResultOnExpiry(true);
        return aggregator;
    }

    // Kinesis 批量发送处理器
    @Bean
    public MessageHandler kinesisOutboundAdapter() {
        KinesisMessageHandler handler = new KinesisMessageHandler(kinesisClient, "your-target-kinesis-stream");
        handler.setOutputChannel(kinesisBatchChannel());
        // 可选：自定义分区键，比如用消息里的ID字段
        // handler.setPartitionKeyExpression(new SpelExpressionParser().parseExpression("payload.id"));
        return handler;
    }
}

额外注意事项 #

• Kinesis单批次支持最大500条或5MB数据，25条的设置完全符合AWS的限制
• 如果需要保证消息顺序，要确保同一来源的消息被分到同一聚合组（Spring Integration默认会基于消息的correlationId分组，拆分器生成的消息默认会继承源文件的关联ID）
• 可以根据业务需求调整批量大小和超时时间，平衡成本和消息延迟

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