使用聚合时无法从JDBC重轮询失败消息的问题及解决诉求

• 现有两个Spring Integration工作流：
  1. 第一个集成流：从SFTP轮询文件 → 消息处理器 → 将消息存入JDBC消息存储的Group1分组
  2. 第二个集成流：从Group1轮询消息 → 聚合器（采用TimeoutCountSequenceSizeReleaseStrategy策略，按阈值/超时批量释放消息）→ 输出处理器处理聚合后的消息组
• 当前问题：当Group1中有2条消息，第二个集成流轮询这2条消息后，若输出处理器发生异常，重轮询仅能获取第二条消息，第一条消息已从JDBC中被删除
• 期望效果：处理失败时，所有被轮询的消息都能重新被轮询；处理成功则从JDBC中删除这些消息

当前核心代码 #

第一个集成流 #

.handle("returns messages as File Info Objects")
.channel(c -> c.queue(jdbcChannelMessageStore, "Group1"))

第二个集成流 #

return IntegrationFlow.from(() -> (Message) jdbcChannelMessageStore.pollMessageFromGroup("Group1"), 
                           e -> e.poller(Pollers.fixedDelay(180000).transactional()))
        .aggregate(a -> a.correlationStrategy(m -> 1)
                          .releaseStrategy(new TimeoutCountSequenceSizeReleaseStrategy(10, 60000))
                          .expireGroupsUponCompletion(true)
                          .outputProcessor("Processes the aggregated messages"));

问题根源与解决办法 #

问题根源 #

直接调用jdbcChannelMessageStore.pollMessageFromGroup()会立即删除消息，即便后续聚合或处理失败，已被轮询的消息也无法回滚。虽然轮询器配置了transactional()，但pollMessageFromGroup是作为消息源的Supplier在事务外执行的，事务仅覆盖后续的聚合和处理步骤，无法回滚已删除的消息。

解决步骤 #

1. 替换自定义消息源，使用JDBC队列通道作为集成流起点
   不要手动调用pollMessageFromGroup，改用JdbcMessageStore支持的队列通道作为第二个集成流的输入，这样轮询操作会被纳入事务管理，处理失败时消息会自动回滚到队列中。

   修改后的第二个集成流代码：

   return IntegrationFlow.from(MessageChannels.queue(jdbcChannelMessageStore, "Group1"), 
                              e -> e.poller(Pollers.fixedDelay(180000).transactional()))
           .aggregate(a -> a.correlationStrategy(m -> 1)
                             .releaseStrategy(new TimeoutCountSequenceSizeReleaseStrategy(10, 60000))
                             .expireGroupsUponCompletion(true)
                             .outputProcessor("Processes the aggregated messages"));
   

2. 事务覆盖完整流程的效果
   轮询器的transactional()配置会将消息取出、聚合、输出处理整个过程纳入同一个事务：

   • 处理成功：事务提交，消息从Group1中永久删除
   • 处理失败（含输出处理器异常）：事务回滚，所有被取出的消息重新回到Group1队列，等待下一次轮询

3. 额外注意事项

   • 确保JdbcMessageStore关联的数据源配置了正确的事务管理器
   • 当前expireGroupsUponCompletion(true)的配置符合需求，处理完成后分组会被自动清理，无需保留状态

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