Spring Integration的TimeoutCountSequenceSizeReleaseStrategy超时未释放消息问题

你遇到的核心问题是聚合器的超时释放策略未生效，主要由以下几个配置错误导致：

1. 错误使用release-strategy-expression属性 #

release-strategy-expression用于编写SpEL表达式（比如你测试用的size() ==1），而直接引用ReleaseStrategy类型的Bean时，应该使用release-strategy属性。原配置用错了属性，导致timeoutReleaseStrategy这个Bean没有被正确加载为释放策略。

2. 缺少超时检查所需的调度器 #

聚合器的超时释放逻辑依赖调度器定期扫描聚合组的超时状态，未配置task-scheduler时，超时触发机制不会生效。

3. 语法错误：correlation-strategy-expression缺少闭合引号 #

原配置中correlation-strategy-expression="headers['type']未闭合引号，会导致配置解析异常，无法按type消息头正确分组。

修正后的完整配置 #

第一步：定义任务调度器 #

<bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
    <property name="poolSize" value="5"/>
</bean>

第二步：修正聚合器与释放策略配置 #

<int:aggregator 
      output-channel="splitterInChannel"
      send-partial-result-on-expiry="true"
      message-store="customMessageStore"
      correlation-strategy-expression="headers['type']" <!-- 补上缺失的闭合引号 -->
      ref="taskAggregationService"
      method="aggregateTask"
      release-strategy="timeoutReleaseStrategy" <!-- 替换为正确的属性 -->
      task-scheduler="taskScheduler" <!-- 添加调度器触发超时检查 -->
      expire-groups-upon-completion="true"
      expire-groups-upon-timeout="true"
      id="customAggregator"
      input-channel="aggregatorInChannel" >
</int:aggregator>

<bean id="customMessageStore"  class="org.springframework.integration.store.SimpleMessageStore"/>
<bean id="timeoutReleaseStrategy" class="org.springframework.integration.aggregator.TimeoutCountSequenceSizeReleaseStrategy">
   <!-- 若希望仅超时触发释放，可将threshold设为极大值，比如Integer.MAX_VALUE -->
   <constructor-arg name="threshold" value="#{T(java.lang.Integer).MAX_VALUE}"/>
   <constructor-arg name="timeout" value="120000"/> <!-- 2分钟 -->
</bean>

额外说明 #

• TimeoutCountSequenceSizeReleaseStrategy的逻辑是：当聚合组消息数达到threshold 或 超时时间到达时，触发释放。如果你的需求是仅在2分钟后释放所有累积消息，将threshold设为Integer.MAX_VALUE即可避免因消息数量达标提前释放。
• 确保SimpleMessageStore的生命周期与应用一致（内存存储仅适用于单实例场景，集群环境需改用分布式消息存储）。

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