Spring Integration动态启动Flow时如何捕获异步异常？

我有一个Spring Integration应用，使用IntegrationFlowContext管理集成流，需要通过暴露的端点对其进行动态启动、停止和重启。现在遇到的问题是：调用integrationFlowContext.getRegistrationById(flowId).start()时抛出的异常（比如文件未找到、队列名称错误、WebSocket URL无效等）无法捕获——我试过用try-catch包裹这段代码，但Lifecycle调用是异步执行的，根本抓不到异常。

补充配置示例 #

以下是包含异常重试逻辑的流配置：

@Configuration
class FlowConfig {
  @Value("${ws.url}")
  private String wsUrl;

  @Bean
  MessageChannel myWebSocketOutputChannel() {
    return new DirectChannel();
  }

  @Bean
  MessageChannel myWebSocketErrorChannel() {
    return new DirectChannel();
  }

  @Bean
  WebSocketClient standardWebSocketClient() {
    return new StandardWebSocketClient();
  }

  @Bean
  IntegrationWebSocketContainer myIntegrationWebSocketContainer(
      final WebSocketClient webSocketClient) throws IOException {
    return new ClientWebSocketContainer(webSocketClient, wsUrl);
  }

  @Bean
  WebSocketInboundChannelAdapter myWebSocketInboundChannelAdapter(
      final IntegrationWebSocketContainer myIntegrationWebSocketContainer,
      final MessageChannel myWebSocketOutputChannel,
      final MessageChannel myWebSocketErrorChannel) {
    final WebSocketInboundChannelAdapter adapter =
        new WebSocketInboundChannelAdapter(myIntegrationWebSocketContainer);
    adapter.setAutoStartup(false);
    adapter.setOutputChannel(myWebSocketOutputChannel);
    adapter.setErrorChannel(myWebSocketErrorChannel);
    return adapter;
  }

  @Bean
  IntegrationFlow myIntegrationFlow(
      final WebSocketInboundChannelAdapter myWebSocketInboundChannelAdapter,
      final MySocketHandler mySocketHandler,
      final MyOutputHandler myOutputHandler) {
    return IntegrationFlow.from(myWebSocketInboundChannelAdapter)
        .<String>handle(mySocketHandler)
        .<MyMessage>filter(Objects::nonNull)
        .handle(myOutputHandler)
        .get();
  }
}

@Component
public class MyWebSocketErrorHandler {
  @ServiceActivator(inputChannel = "myWebSocketErrorChannel")
  public void handleError(final Message<?> message) {
    log.error("WebSocket error: {}", message.getPayload());
  }
}

比如启动流时WebSocket不可用，我看不到自定义的错误日志，只能看到DeploymentException的堆栈跟踪。MySocketHandler和MyOutputHandler是简单的GenericHandler实现，此处省略。

异常堆栈信息 #

启动流时WebSocket URL不可用的异常日志如下：

2023-10-25T18:47:55.304Z  INFO 1 --- [nio-8811-exec-1] c.m.r.i.flow.IntegrationFlowManager      : Starting integration flow MY
2023-10-25T18:47:55.304Z  INFO 1 --- [nio-8811-exec-1] er$IntegrationWebSocketConnectionManager : Starting IntegrationWebSocketConnectionManager
2023-10-25T18:47:55.304Z  INFO 1 --- [nio-8811-exec-1] er$IntegrationWebSocketConnectionManager : Connecting to WebSocket at wss://ws-simulator:3001/ws
2023-10-25T18:47:55.313Z  INFO 1 --- [nio-8811-exec-1] o.s.i.w.i.WebSocketInboundChannelAdapter : started bean 'MY.websocket:inbound-channel-adapter#0'
2023-10-25T18:47:55.410Z ERROR 1 --- [cTaskExecutor-1] er$IntegrationWebSocketConnectionManager : Failed to connect

jakarta.websocket.DeploymentException: The HTTP request to initiate the WebSocket connection to [wss://ws-simulator:3001/ws] failed
at org.apache.tomcat.websocket.WsWebSocketContainer.connectToServerRecursive(WsWebSocketContainer.java:429) ~[tomcat-embed-websocket-10.1.11.jar!/:na]
at org.apache.tomcat.websocket.WsWebSocketContainer.connectToServer(WsWebSocketContainer.java:179) ~[tomcat-embed-websocket-10.1.11.jar!/:na]
at org.springframework.web.socket.client.standard.StandardWebSocketClient.lambda$executeInternal$0(StandardWebSocketClient.java:150) ~[spring-websocket-6.0.11.jar!/:6.0.11]
at org.springframework.util.concurrent.FutureUtils.lambda$toSupplier$0(FutureUtils.java:74) ~[spring-core-6.0.11.jar!/:6.0.11]
at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
Caused by: java.util.concurrent.ExecutionException: java.io.EOFException: Unexpected end of stream
at org.apache.tomcat.websocket.AsyncChannelWrapperSecure$WrapperFuture.get(AsyncChannelWrapperSecure.java:492) ~[tomcat-embed-websocket-10.1.11.jar!/:na]
at org.apache.tomcat.websocket.WsWebSocketContainer.processResponse(WsWebSocketContainer.java:793) ~[tomcat-embed-websocket-10.1.11.jar!/:na]
at org.apache.tomcat.websocket.WsWebSocketContainer.connectToServerRecursive(WsWebSocketContainer.java:335) ~[tomcat-embed-websocket-10.1.11.jar!/:na]
... 5 common frames omitted
Caused by: java.io.EOFException: Unexpected end of stream
at org.apache.tomcat.websocket.AsyncChannelWrapperSecure$ReadTask.run(AsyncChannelWrapperSecure.java:261) ~[tomcat-embed-websocket-10.1.11.jar!/:na]        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
... 1 common frames omitted

针对Spring Integration异步启动时的异常捕获问题，分以下场景处理：

1. WebSocket连接异常的专属处理 #

你配置的errorChannel仅处理消息流转过程中的异常，但WebSocket连接由IntegrationWebSocketConnectionManager异步发起，这类启动阶段的连接异常不会进入消息流的错误通道。需要给ClientWebSocketContainer设置自定义ConnectionListener捕获连接失败事件：

@Bean
IntegrationWebSocketContainer myIntegrationWebSocketContainer(
    final WebSocketClient webSocketClient,
    final ApplicationContext applicationContext) throws IOException {
  ClientWebSocketContainer container = new ClientWebSocketContainer(webSocketClient, wsUrl);
  container.setConnectionListener(new ConnectionListener() {
    @Override
    public void onConnect(WebSocketSession session) {
      // 连接成功的自定义逻辑
    }

    @Override
    public void onDisconnect(WebSocketSession session) {
      // 连接断开的自定义逻辑
    }

    @Override
    public void onFailure(Throwable ex) {
      // 捕获连接失败异常，发送到自定义错误通道
      MessageChannel errorChannel = applicationContext.getBean("myWebSocketErrorChannel", MessageChannel.class);
      errorChannel.send(MessageBuilder.withPayload(ex).build());
    }
  });
  return container;
}

这样连接失败时，异常会被onFailure方法捕获，进而发送到myWebSocketErrorChannel，MyWebSocketErrorHandler就能收到并打印自定义日志。

2. 通用集成流启动异常捕获 #

对于文件适配器、队列连接等其他组件的异步启动异常，可通过以下方式处理：

• 自定义LifecycleProcessor：替换Spring默认的DefaultLifecycleProcessor，重写startBeans方法，在异步执行启动逻辑时添加异常捕获。
• 监听Spring事件：通过ApplicationListener捕获LifecycleEvent，或者监听集成流注册状态变化的自定义事件，从中提取启动异常。
• 设置线程池异常处理器：给执行启动逻辑的线程池配置UncaughtExceptionHandler，捕获异步线程中的未处理异常。

3. 同步启动集成流（可选） #

如果业务允许同步执行启动逻辑（无需异步提升性能），可修改集成流注册方式，让启动过程同步执行，此时异常可被try-catch捕获：

IntegrationFlowRegistration registration = integrationFlowContext.registration(myIntegrationFlow)
    .addBean(myWebSocketInboundChannelAdapter)
    .register();
// 同步启动，异常可直接捕获
try {
  registration.start();
} catch (Exception e) {
  log.error("启动集成流失败", e);
}

注意：WebSocket连接默认是异步操作，即使同步调用start()，连接逻辑仍会在后台线程执行，因此WebSocket场景仍需配合ConnectionListener处理异常。

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