Spring Integration 应用重启后，持久化在数据库中的未完成消息组（如因超时未释放的聚合组）将无法自动触发发送；需通过 setExpireTimeout() 启用孤儿组清理机制，并配合 purgeOrphanedGroups() 在启动时主动扫描并释放过期组。
spring integration 应用重启后，持久化在数据库中的未完成消息组（如因超时未释放的聚合组）将无法自动触发发送；需通过 `setexpiretimeout()` 启用孤儿组清理机制，并配合 `purgeorphanedgroups()` 在启动时主动扫描并释放过期组。
在基于 JdbcMessageStore 的 Spring Integration 聚合场景中，若应用在消息组尚未超时释放前意外重启，这些“孤儿组”（orphaned groups）——即数据库中存在、但内存中无对应聚合器上下文的 MessageGroup——将长期滞留，既不被释放，也不触发 releaseStrategy 或 groupTimeoutExpression 对应的处理逻辑。默认情况下，MessageGroupStoreReaper 并不会在应用启动时自动执行清理，它仅作为可调度组件，需显式调用或由定时任务驱动。
✅ 正确解决方案：启用 expireTimeout（推荐，Spring Integration 5.4+）
自 5.4 版本起，AggregatingMessageHandler 原生支持启动时自动清理孤儿组，只需配置 expireTimeout 即可：
@Bean
public MessageHandler aggregator() {
AggregatingMessageHandler aggregator =
new AggregatingMessageHandler(
new DefaultAggregatingMessageGroupProcessor(),
jdbcMessageStore()
);
// ... 其他配置（correlationStrategy, releaseStrategy 等）
// ⚠️ 关键：设置非零 expireTimeout（单位：毫秒）
// 应用启动时将自动调用 purgeOrphanedGroups() 扫描并释放所有 timestamp <blockquote><p>✅ setExpireTimeout(10_000L) 表示：应用启动时，立即清理所有最后更新时间早于当前时间 10 秒以上的消息组（即已“超时”的孤儿组）。这直接解决重启后积压消息无法发送的核心问题。</p></blockquote><h3>? 补充说明与注意事项</h3>
expireTimeout ≠ groupTimeoutExpression
groupTimeoutExpression 控制运行时组的超时释放逻辑；而 expireTimeout 是启动期专用机制，专为恢复持久化存储中“断连状态”的组设计，二者互补，建议同时配置。-
purgeOrphanedGroups() 可手动触发
若需更精细控制（如延迟启动清理、或结合健康检查），可注入 AggregatingMessageHandler 并手动调用：@PostConstruct public void initAfterStartup() { try { Thread.sleep(2000); // 确保 JDBC store 已就绪 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } aggregator.purgeOrphanedGroups(); // 主动清理 } 关于 MessageGroupStoreReaper 的常见误区
如文档明确指出：MessageGroupStoreReaper 本身不自动运行，必须由 @Scheduled 方法显式调用其 run() 方法（如提问者 UPD 中的方案）。但该方式属于“兜底轮询”，不如 expireTimeout 启动即治来得精准高效。数据库兼容性要求
purgeOrphanedGroups() 依赖 JdbcMessageStore 的 removeMessageGroup() 和 getMessageGroupIds() 实现，确保所用数据库支持标准 SQL DELETE ... WHERE timestamp
✅ 最终建议配置（精简可靠）
@Bean
public MessageHandler aggregator() {
AggregatingMessageHandler aggregator =
new AggregatingMessageHandler(
new DefaultAggregatingMessageGroupProcessor(),
jdbcMessageStore()
);
aggregator.setCorrelationStrategy(new MyCorrelationStrategy());
aggregator.setReleaseStrategy(messageGroup ->
System.currentTimeMillis() - messageGroup.getTimestamp() >= 10_000L);
aggregator.setGroupTimeoutExpression(new ValueExpression(10_000L));
aggregator.setSendPartialResultOnExpiry(true);
aggregator.setExpireGroupsUponTimeout(true);
// ✅ 核心：启动时自动清理已超时的孤儿组
aggregator.setExpireTimeout(10_000L);
// ✅ 可选：每30秒周期性检查（增强鲁棒性）
aggregator.setExpireDuration(30_000L);
return aggregator;
}
并确保主类启用定时任务支持：
@EnableScheduling
@SpringBootApplication
public class SpringIntegrationExampleApplication { /* ... */ }
通过以上配置，应用每次重启后将自动识别并释放所有已超时的持久化消息组，确保消息不丢失、业务逻辑不中断，真正实现高可用聚合流程。
