Spring Integration Java DSL如何为不同并行流配置自定义超时

核心实现逻辑 #

你当前使用的scatterGather组件本身支持超时配置，结合Spring Integration提供的请求处理器超时切面，可以给每个子流单独设置超时值，同时支持配置文件动态修改，不需要硬编码。 #

具体实现步骤 #

1. 新增可配置超时参数 #

在application.yml/application.properties中定义各流程的超时值，支持按环境调整：

integration:
  timeout:
    global-gather: 5000 # 聚合器最大等待超时，单位毫秒
    flow1: 1000 # 保存请求流程超时
    flow2: 3000 # 数据查询流程超时
    flow3: 2000 # CD请求生成流程超时

2. 注入配置值，定义超时切面 #

在IntegrationConfiguration配置类中注入配置参数，为每个子流定义独立的超时通知Bean：

// 注入配置的超时值，设置默认兜底值
@Value("${integration.timeout.flow1:1000}")
private long flow1Timeout;
@Value("${integration.timeout.flow2:3000}")
private long flow2Timeout;
@Value("${integration.timeout.flow3:2000}")
private long flow3Timeout;
@Value("${integration.timeout.global-gather:5000}")
private long globalGatherTimeout;

// 各子流独立超时切面
@Bean
public RequestHandlerTimeoutAdvice flow1TimeoutAdvice() {
    RequestHandlerTimeoutAdvice advice = new RequestHandlerTimeoutAdvice();
    advice.setTimeout(flow1Timeout);
    // 超时后返回兜底标识，若需直接抛异常可关闭该配置
    advice.setReturnFailureResultIfException(true);
    advice.setFailureResult("flow1:处理超时");
    return advice;
}

@Bean
public RequestHandlerTimeoutAdvice flow2TimeoutAdvice() {
    RequestHandlerTimeoutAdvice advice = new RequestHandlerTimeoutAdvice();
    advice.setTimeout(flow2Timeout);
    advice.setReturnFailureResultIfException(true);
    advice.setFailureResult("flow2:处理超时");
    return advice;
}

@Bean
public RequestHandlerTimeoutAdvice flow3TimeoutAdvice() {
    RequestHandlerTimeoutAdvice advice = new RequestHandlerTimeoutAdvice();
    advice.setTimeout(flow3Timeout);
    advice.setReturnFailureResultIfException(true);
    advice.setFailureResult("flow3:处理超时");
    return advice;
}

3. 修改子流绑定超时切面 #

给每个子流的消息处理器绑定对应的超时切面，注意原有代码中子流处理器无返回值，需补充返回值否则聚合后拿不到处理结果：

// flow1示例，flow2、flow3逻辑一致替换对应advice即可
@Bean
public IntegrationFlow flow1() {
  return integrationFlowDefination ->
      integrationFlowDefination
          .channel(c -> c.executor(Executors.newCachedThreadPool()))
          .handle(
              message -> {
                try {
                  lionService.saveLionRequest(
                      (LionRequest) message.getPayload(), String.valueOf(dbId));
                  return "flow1:处理成功";
                } catch (JsonProcessingException e) {
                  throw new RuntimeException(e);
                }
              },
              // 绑定该流专属的超时切面
              handler -> handler.advice(flow1TimeoutAdvice())
          );
}

4. 给聚合器配置全局兜底超时 #

修改主流程的scatterGather配置，增加全局聚合超时，避免极端情况整体阻塞：

.scatterGather(
    scatterer ->
        scatterer
            .applySequence(true)
            .recipientFlow(flow1())
            .recipientFlow(flow2())
            .recipientFlow(flow3()),
    gatherer -> gatherer
        .releaseLockBeforeSend(true)
        .timeout(globalGatherTimeout) // 到最大等待时间直接释放已收到的子流结果
)

原有代码问题修正 #

• 不要在生产环境使用Executors.newCachedThreadPool()，该线程池无最大线程数限制，高并发下易触发OOM，建议替换为你已经定义的ThreadPoolTaskExecutor，给每个子流配置合理的线程数、队列长度。
• 配置类中定义的long dbId = new SequenceGenerator().nextId();只会在配置类初始化时执行一次，所有请求会共用同一个dbId，如果需要每个请求生成独立ID，要把ID生成逻辑移到子流的处理方法内。
• Controller层的System.out.Println存在拼写错误，正确写法是System.out.println，生产环境建议使用日志框架打印，不要直接用控制台输出。
• 如果超时后不需要返回兜底值、需要直接中断请求，把超时切面的setReturnFailureResultIfException配置为false，再搭配全局异常捕获返回超时错误即可。

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