Spring Integration聚合器与SimpleMessageGroup内存泄漏问题求助

你在Spring Integration 4.3.10版本的批处理场景中遇到的SimpleMessageGroup内存泄漏问题，我结合你的配置尝试和现象来帮你定位并解决：

问题核心原因 #

你当前使用的SimpleMessageStore是最基础的内存消息存储实现，在4.3.x版本中它的过期分组清理逻辑存在局限性——即使你配置了expireGroupsUponCompletion和expireGroupsUponTimeout，分组对象也可能因为清理触发时机不及时、或内部引用未被正确释放而残留。另外，早期版本的注解式聚合器配置，部分参数的生效优先级可能和预期不一致，导致清理规则没有真正落地。

分步解决方案 #

1. 替换内存消息存储实现 #

把SimpleMessageStore换成DefaultMessageStore，它提供了更可靠的过期分组管理和清理机制：

@Bean
public MessageStore messageStore() {
    DefaultMessageStore messageStore = new DefaultMessageStore();
    // 直接设置分组超时时间，确保超期分组能被识别
    messageStore.setGroupTimeout(1000L);
    return messageStore;
}

2. 优化AggregatorFactoryBean配置 #

在你的工厂Bean配置中，补充关键参数并确保协同生效：

@Bean
@ServiceActivator(inputChannel = "serviceResponseChannel")
FactoryBean<MessageHandler> aggregatorFactoryBean(MessageStore messageStore) {
    AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
    
    aggregatorFactoryBean.setProcessorBean(new MyAggregator(myTransactionLogger()));
    aggregatorFactoryBean.setMethodName("aggregate");
    aggregatorFactoryBean.setMessageStore(messageStore); // 使用新的MessageStore
    aggregatorFactoryBean.setOutputChannel(aggregatorResponseChannel());
    
    aggregatorFactoryBean.setExpireGroupsUponTimeout(true);
    aggregatorFactoryBean.setGroupTimeoutExpression(new ValueExpression<>(1900L));
    aggregatorFactoryBean.setSendPartialResultOnExpiry(false);
    aggregatorFactoryBean.setExpireGroupsUponCompletion(true);
    
    // 新增：强制移除已取消的分组，避免残留
    aggregatorFactoryBean.setRemoveCanceledGroups(true);
    // 显式配置关联策略，确保分组按预期关联（默认也是correlationId，显式配置更清晰）
    aggregatorFactoryBean.setCorrelationStrategy(new HeaderAttributeCorrelationStrategy("correlationId"));
    // 配置释放策略，确保分组满足条件后立即释放
    aggregatorFactoryBean.setReleaseStrategy(new SequenceSizeReleaseStrategy());
    
    return aggregatorFactoryBean;
}

3. 添加主动定时清理任务 #

Spring Integration 4.3.x中，DefaultMessageStore的清理默认是被动触发的（比如有新消息进入时），如果你的批处理场景存在空闲期，残留分组可能无法及时被清理。添加定时任务主动触发清理：

@Autowired
private MessageStore messageStore;

@Scheduled(fixedRate = 5000) // 每5秒执行一次清理
public void cleanExpiredGroups() {
    messageStore.expireMessageGroups();
}

4. 检查自定义聚合方法 #

确认你的aggregate方法中没有持有Message或SimpleMessageGroup的长期引用——比如不要把这些对象存入静态变量、全局缓存或其他生命周期长的容器中，避免GC无法回收这些对象。

验证手段 #

• 重新运行测试，完成后获取堆转储，检查SimpleMessageGroup对象是否被正常回收
• 监控JVM堆内存变化，确认高吞吐量场景下内存不再持续攀升

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