本文介绍如何通过配置 setExpireTimeout() 和合理使用 MessageGroupStoreReaper，使 Spring Integration 在应用重启后自动清理并释放数据库中滞留的过期消息组，确保未完成聚合的消息能被及时处理或丢弃。
本文介绍如何通过配置 `setexpiretimeout()` 和合理使用 `messagegroupstorereaper`，使 spring integration 在应用重启后自动清理并释放数据库中滞留的过期消息组，确保未完成聚合的消息能被及时处理或丢弃。
在基于 Spring Integration 的持久化聚合场景中（如使用 JdbcMessageStore），应用意外重启会导致内存中正在等待超时的 MessageGroup 丢失，而其对应的消息仍滞留在数据库中——这些“孤儿组”（orphaned groups）既不会被自动释放，也不会触发 ReleaseStrategy 或超时逻辑，从而造成消息积压与业务逻辑中断。
Spring Integration 自 5.4 版本起 提供了原生支持来解决该问题：通过 AggregatingMessageHandler#setExpireTimeout(long) 方法，可启用启动时及周期性扫描，主动识别并清理数据库中超过指定时限、且无新消息到达的孤儿组。
✅ 正确配置方式（推荐）
只需在 AggregatingMessageHandler 初始化时设置 expireTimeout（单位：毫秒），框架将自动在应用启动时执行一次 purgeOrphanedGroups()，并按 expireDuration（默认为 expireTimeout）周期性调度清理任务：
@Bean
@ServiceActivator(inputChannel = "consumingChannel")
public MessageHandler aggregator() {
AggregatingMessageHandler aggregator =
new AggregatingMessageHandler(
new DefaultAggregatingMessageGroupProcessor(),
jdbcMessageStore()
);
// 关键：启用孤儿组自动清理（例如：10秒）
aggregator.setExpireTimeout(10_000L);
// 其他原有配置保持不变
aggregator.setGroupTimeoutExpression(new ValueExpression<>(10_000L));
aggregator.setCorrelationStrategy(new MyCorrelationStrategy());
aggregator.setSendPartialResultOnExpiry(true);
aggregator.setExpireGroupsUponTimeout(true);
aggregator.setReleaseStrategy(group ->
System.currentTimeMillis() - group.getTimestamp() >= 10_000L
);
return aggregator;
}⚠️ 注意：setExpireTimeout() 是最简、最可靠的方案；它内部已集成 MessageGroupStoreReaper 的调用逻辑，并与 ForceReleaseProcessor 协同工作（如 sendPartialResultOnExpiry=true 时会强制发布部分结果）。
❌ 不推荐的手动调度方式（如原答案中的 @Scheduled）
虽然可通过 @EnableScheduling + @Scheduled 主动调用 reaper.run() 实现类似效果，但存在明显缺陷：
- MessageGroupStoreReaper 默认不启用 forceReleaseProcessor，无法触发 sendPartialResultOnExpiry 或 discardChannel；
- 手动调用 reaper.run() 属于低层操作，绕过了 Aggregator 的语义控制，易导致状态不一致；
- fixedDelay = Long.MAX_VALUE 实际仅执行一次，违背“周期性清理”设计初衷，且不可靠。
因此，请优先使用 setExpireTimeout()，而非自行调度 Reaper。
? 补充说明
- expireTimeout 的含义是：若某 MessageGroup 自最后更新时间起，已超过该毫秒数且无新消息加入，则视为孤儿组并强制过期；
- 清理过程会调用 ForceReleaseProcessor（即 Aggregator 自身的 releaseStrategy 和 outputChannel），确保 sendPartialResultOnExpiry、discardChannel 等策略生效；
- 数据库表（如 MESSAGE_GROUP）中 LAST_RELEASED_SEQUENCE 和 CREATED_DATE 字段会被用于判断时效性，请确保数据库时钟准确；
- 若需自定义过期判定逻辑（如按业务时间而非系统时间），可继承 DefaultMessageGroupStore 并重写 removeOrphans() 方法，但通常无需如此。
✅ 总结
| 方案 | 是否推荐 | 说明 |
|---|---|---|
| aggregator.setExpireTimeout(10_000L) | ✅ 强烈推荐 | 开箱即用，语义清晰，与聚合器生命周期深度集成 |
| @Scheduled 调用 reaper.run() | ❌ 不推荐 | 绕过核心机制，易出错，维护成本高 |
| 仅依赖 MessageGroupStoreReaper（未配置 expireTimeout） | ❌ 无效 | Reaper 默认不执行 forceRelease，仅删除空组 |
启用 setExpireTimeout() 后，应用每次重启都会自动扫描并释放滞留的过期消息组，真正实现“故障恢复后消息不丢失、不卡死”的高可用聚合能力。
