如何为每个Poller轮询周期的消息注入唯一correlation-id实现聚合？

核心思路 #

给每一轮Poller获取到的所有消息打上同一个唯一Correlation ID，让聚合器能识别出这些属于同一批次的消息，进而完成聚合后再通过FTP发送。

具体实现方案 #

1. 自定义批次ID注入逻辑 #

在Poller获取消息后、进入聚合器前，添加一个消息转换器，为当前轮询批次的所有消息注入相同的Correlation ID。利用ThreadLocal存储当前批次ID，结合Spring Integration的@AfterPoll钩子在批次结束后清理，避免线程复用导致的ID混乱：

@Component
public class BatchCorrelationInjector {
    private final ThreadLocal<String> batchIdHolder = new ThreadLocal<>();

    @Transformer(inputChannel = "polledSourceChannel", outputChannel = "aggregatorInputChannel")
    public Message<String> injectBatchId(Message<String> originalMsg) {
        // 首次处理当前批次消息时生成唯一ID
        if (batchIdHolder.get() == null) {
            batchIdHolder.set(UUID.randomUUID().toString());
        }
        // 复制原消息并添加Correlation ID头
        return MessageBuilder.fromMessage(originalMsg)
                .setHeader(MessageHeaders.CORRELATION_ID, batchIdHolder.get())
                .build();
    }

    // 轮询批次结束后清理ThreadLocal，防止内存泄漏
    @AfterPoll
    public void cleanBatchId() {
        batchIdHolder.remove();
    }
}

2. 配置聚合器匹配批次ID #

聚合器需配置为按Correlation ID分组，同时设置释放策略匹配你的业务需求（最多m条/轮询周期结束）：

@Bean
public AggregatorFactoryBean batchAggregator() {
    AggregatorFactoryBean aggregator = new AggregatorFactoryBean();
    aggregator.setInputChannel("aggregatorInputChannel");
    aggregator.setOutputChannel("ftpSendChannel");
    
    // 按消息头中的Correlation ID分组
    aggregator.setCorrelationStrategy(msg -> msg.getHeaders().get(MessageHeaders.CORRELATION_ID));
    // 达到m条消息时释放聚合结果
    aggregator.setReleaseStrategy(new MessageCountReleaseStrategy(m));
    // 轮询周期结束后强制释放未达m条的批次，避免消息积压
    aggregator.setSendPartialResultOnExpiry(true);
    aggregator.setGroupTimeout(n * 1000L); // 与轮询周期n秒保持一致
    return aggregator;
}

3. 配置Poller的批次上限 #

确保Poller的maxMessagesPerPoll设置为m，让每轮最多获取m条消息，和聚合器的释放策略对齐：

@Bean
public PollerMetadata batchPoller() {
    PollerMetadata poller = new PollerMetadata();
    poller.setTrigger(new PeriodicTrigger(n * 1000L)); // 每n秒轮询一次
    poller.setMaxMessagesPerPoll(m); // 每轮最多获取m条消息
    return poller;
}

关键注意事项 #

• 必须通过@AfterPoll清理ThreadLocal中的批次ID，否则线程池复用会导致不同批次的消息被打上相同ID，聚合逻辑混乱。
• 聚合器的groupTimeout要和Poller的轮询周期一致，确保即使批次未达到m条，到时间也会释放聚合结果，避免消息长期积压。
• 如果原消息本身带有Correlation ID，可根据业务需求添加判断逻辑，决定是否覆盖原ID。

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