Spring Integration：启动通道适配器后等待流程完成再返回REST响应

要实现REST请求等待整个集成流程完成后再返回，核心是让请求线程阻塞等待流程结束的信号（即taskStatusChannel中的COMPLETED消息），具体实现步骤如下：

1. 调整RestController逻辑 #

在REST接口中启动JDBC轮询适配器后，通过QueueChannel的阻塞接收方法等待流程完成信号，并处理超时、异常场景：

@RestController
@RequestMapping("/flow")
public class FlowTriggerController {

    private final MessageChannel controlChannel;
    private final QueueChannel taskStatusChannel;
    private final JdbcPollingChannelAdapter jdbcPollingAdapter;
    private static final long WAIT_TIMEOUT = 30000; // 30秒超时

    public FlowTriggerController(MessageChannel controlChannel, QueueChannel taskStatusChannel,
                                 JdbcPollingChannelAdapter jdbcPollingAdapter) {
        this.controlChannel = controlChannel;
        this.taskStatusChannel = taskStatusChannel;
        this.jdbcPollingAdapter = jdbcPollingAdapter;
    }

    @PostMapping("/start")
    public ResponseEntity<String> startFullFlow() {
        // 初始化状态：确保GraphQL适配器处于停止状态，避免残留任务干扰
        controlChannel.send(new GenericMessage<>("@graphQLPollingAdapter.stop()"));
        
        // 清空状态通道的残留消息
        while (taskStatusChannel.receive(0) != null) {}

        // 启动JDBC轮询适配器
        jdbcPollingAdapter.start();

        try {
            // 阻塞等待流程完成信号
            Message<?> statusMsg = taskStatusChannel.receive(WAIT_TIMEOUT);
            if (statusMsg != null && "COMPLETED".equals(statusMsg.getPayload())) {
                return ResponseEntity.ok("全流程执行完成");
            } else {
                // 超时或未收到完成信号，强制停止所有适配器
                stopAllAdapters();
                return ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
                        .body("流程执行超时，未完成");
            }
        } catch (Exception e) {
            // 异常场景，停止所有适配器并返回错误
            stopAllAdapters();
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body("流程执行异常：" + e.getMessage());
        }
    }

    private void stopAllAdapters() {
        controlChannel.send(new GenericMessage<>("@jdbcPollingAdapter.stop()"));
        controlChannel.send(new GenericMessage<>("@graphQLPollingAdapter.stop()"));
    }
}

2. 确保taskStatusChannel配置正确 #

确认taskStatusChannel是容量合适的QueueChannel，避免消息堆积：

@Bean
public QueueChannel taskStatusChannel() {
    return new QueueChannel(1); // 仅需存储一条完成信号，容量设为1即可
}

3. 优化集成流的可靠性 #

• 在flow1中，确保过滤通过后再切换适配器，避免无效触发：

  @Bean
  public IntegrationFlow flow1() {
      return IntegrationFlow.from(eventChannel)
                            .filter(eventFilter, f -> f.discardChannel(discardChannel))
                            .handle((msg, headers) -> {
                                trigger("@jdbcPollingAdapter.stop()");
                                trigger("@graphQLPollingAdapter.start()");
                                return null;
                            })
                            .get();
  }
  

• 在graphQLFlow中，确保获取到目标数据后再发送完成信号：

  @Bean
  public IntegrationFlow graphQLFlow() {
      return IntegrationFlow.from(graphQLChannel)
                            .filter(graphQLFilter, f -> f.discardChannel(discardChannel))
                            .handle((msg, headers) -> {
                                trigger("@graphQLPollingAdapter.stop()");
                                taskStatusChannel.send(new GenericMessage<>("COMPLETED"));
                                return null;
                            })
                            .get();
  }
  

关键说明 #

• QueueChannel.receive(timeout)是阻塞方法，会让REST请求线程暂停，直到收到消息或超时，完美匹配"等待流程完成"的需求。
• 每次请求前清空taskStatusChannel的残留消息，避免上一次流程的信号干扰当前请求。
• 超时和异常场景下强制停止所有适配器，防止资源泄漏。

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