Spring Integration捕获异常后聚合器释放策略异常问题排查

问题 #

我有大量消息被放入QueueChannel中，每个Object会有多个更新（hashCode不同、equals()相同、属性值不同），按时间事件顺序写入。我使用Aggregator作为缓冲区，将消息以集合形式推送。之后，Handler会将最新版本的Object存入Map，再通过JdbcTemplate.batchUpdate()将Map的values()写入数据库。

此流程运行正常，直到batchUpdate()执行失败。捕获异常后，释放策略出现异常：异常发生前，释放策略每秒推送一个包含约百条对象的集合；异常发生后，释放策略每1-2毫秒仅推送一个包含单个对象的集合。

（当前使用Spring Integration 6.2.5与JDK 21）

代码片段：

IntegrationFlow
   .from(myQueueChannel())
   .aggregate(as -> as
     .releaseStrategy(new TimeoutCountSequenceSizeReleaseStrategy(1000, 1000))
     .correlationStrategy(cs -> "buffer")
     .expireGroupsUponCompletion(true)
     .expireGroupsUponTimeout(true)
     .requiresReply(false)
     .messageStore(new SimpleMessageStore())
     .poller(Pollers.fixedRate(100).taskExecutor( // new
         Executors.newSingleThreadExecutor())
     )
   )
   .channel(myExecutorChannel() // new
   .get();
}

@Bean IntegrationFlow writer() {
   return IntegrationFlow
   .from(myExecutorChannel()) // new

   .<List<MyObject>>handle((p, h) -> {
      log.info(p.size() + " All field updates.");
      Map<MyKey, MyObject> last = new HashMap<>();
      for (MyObject o: p) {
         last.put(o.getId(), o);
      }
      try {
         int response[][] = myRepo.myBatchUpdate(new ArrayList<>(last.values()), 100);               
      } catch (Exception e) {
         log.error("Caught ", e);
      }
      return null;
   })
   .get();

请问为何已捕获的异常会影响释放策略？

分析与解答 #

核心原因是异常被Handler内部捕获后，Aggregator的关联组未被正常销毁，导致后续消息的聚合逻辑完全失效：

1. 组状态异常
   你设置了expireGroupsUponCompletion(true)，但这个配置生效的前提是Aggregator收到明确的“处理完成”信号。虽然你在Handler中捕获了batchUpdate的异常，但返回null的操作并没有触发Aggregator的组完成逻辑。此时内存中的SimpleMessageStore里，原关联组的状态会停留在“已释放但未完成”的异常状态。

2. 频繁释放的触发逻辑
   由于关联策略固定返回"buffer"，所有新消息都会进入这个异常状态的组。而TimeoutCountSequenceSizeReleaseStrategy在处理异常状态的组时，会错误判定组满足释放条件——每次新消息加入后，组会被立即释放，导致每条消息单独推送，出现1-2毫秒推送单个对象的情况。

修复方案：

• 确保组处理后无论成败都完成：在Handler中，无论batchUpdate成功还是失败，都要明确触发组的完成。可以通过返回空消息（而非null），告知Aggregator销毁当前组。
  示例修改Handler：

  .<List<MyObject>>handle((p, h) -> {
      log.info(p.size() + " All field updates.");
      Map<MyKey, MyObject> last = new HashMap<>();
      for (MyObject o: p) {
          last.put(o.getId(), o);
      }
      try {
          int response[][] = myRepo.myBatchUpdate(new ArrayList<>(last.values()), 100);               
      } catch (Exception e) {
          log.error("Caught ", e);
      }
      // 返回空消息触发组完成
      return MessageBuilder.withPayload(null).build();
  })
  

• 配置错误通道：给Aggregator配置errorChannel，让异常被统一处理，而非被Handler吞掉。这样Aggregator可以自动处理异常情况下的组销毁逻辑。
• 替换消息存储：如果内存存储的状态管理不够可靠，可以改用JdbcMessageStore等持久化消息存储，避免内存中组状态异常无法恢复的问题。

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