Spring Integration流程中断排查：数据无法从collectFlow流转至processFlow

1. 检查PROCESS_CHANNEL队列状态 #

因为你用的是QueueChannel，消息可能堆积在队列中未被消费：

• 借助Spring Actuator的/integration/channels端点，查看PROCESS_CHANNEL的关键指标：
  • queueSize：当前队列中的消息数
  • remainingCapacity：队列剩余容量
  • sendCount：成功发送到队列的消息总数
  • receiveCount：从队列接收的消息总数
    如果queueSize大于0但receiveCount没有增长，说明消息确实堆积在队列，消费者未正常工作。

2. 验证processFlow的初始化状态 #

检查processFlow是否成功创建并启动：

• 查看应用启动日志，搜索是否有BeanCreationException或与processFlow、processingService相关的初始化错误
• 确认所有processingService1~N的Bean都正常加载，没有依赖缺失或配置错误

3. 捕获消息发送异常 #

collectFlow的最后一步发送消息到PROCESS_CHANNEL时，可能出现未被捕获的异常：

• 在collectFlow的channel(ChannelConfig.PROCESS_CHANNEL)后添加日志，确认消息是否成功发送：

  .log(LoggingHandler.Level.TRACE, logger.getName(), m -> "Message sent to PROCESS_CHANNEL, payload: " + m.getPayload())
  

• 为collectFlow添加错误处理通道，捕获发送阶段的异常：

  .errorChannel("collectErrorChannel")
  

  然后创建错误处理流程监听该通道，打印异常信息：

  @Bean
  public IntegrationFlow collectErrorFlow() {
      return IntegrationFlows.from("collectErrorChannel")
              .log(LoggingHandler.Level.ERROR, logger.getName(), m -> "Failed to send message: " + m.getPayload())
              .get();
  }
  

4. 检查消息合法性 #

确认消息 payload（CollectionContainer）是否存在问题：

• 检查CollectionContainer是否实现了Serializable接口（虽然内存队列默认不需要，但如果有自定义序列化逻辑可能影响）
• 在collectFlow的最后日志中打印payload的详细内容，确认数据结构是否正常，是否有字段为空或不符合processFlow的预期

5. 排查线程阻塞问题 #

如果processFlow的消费者线程被阻塞，会导致无法从队列接收新消息：

• 使用jstack命令导出应用线程栈，查看是否有线程卡在processingService1~N的方法中（比如死锁、长时间IO、无限循环）
• 检查processFlow是否使用了自定义线程池，线程池的核心线程数、队列容量是否足够，是否出现线程耗尽的情况

6. 验证通道引用一致性 #

确认collectFlow发送的通道和processFlow监听的通道是同一个实例：

• 检查ChannelConfig.PROCESS_CHANNEL常量的字符串值是否完全一致，没有拼写错误或大小写差异
• 查看Spring上下文，确认processChannel() Bean是否唯一，没有被其他同名Bean覆盖

7. 启用更详细的日志 #

除了Spring Integration的TRACE日志，补充开启消息模块的TRACE日志：

• 在日志配置文件中添加：

  <logger name="org.springframework.messaging" level="TRACE"/>
  <logger name="org.springframework.integration.channel" level="TRACE"/>
  

  这会输出消息在通道中发送、接收的全链路细节，帮助定位消息是否真的到达PROCESS_CHANNEL，以及是否有消费者尝试处理。

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