Spring Integration如何向网关传多参数并在子流中使用

问题根因 #

• 你通过网关@Header注入的sourceSystem参数，默认会随消息完整透传到整个流链路，Scatter-Gather组件分发消息给各个子流时会自动复制全部请求头，不需要额外做传递配置。
• 之前拿不到header的核心原因是transformer的写法问题：如果lambda只声明payload类型作为入参，框架只会把消息体传入，自然拿不到头信息；只要把入参声明为Message类型，就能访问完整的消息头和消息体。
• 你碰到的this component is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'报错，本质是流中存在无返回值的单向处理端点（比如你flow1结尾的nullChannel是吞掉所有消息不返回的组件），如果误将这类端点配置为需要输出结果给下游聚合器，框架就会抛出该错误。
• 额外注意：你当前代码中long dbId = new SequenceGenerator().nextId();是配置类初始化时执行一次的成员变量，所有请求都会共用同一个dbId，属于明显bug，需要调整到请求处理逻辑中按请求生成。

具体代码修改 #

1. 修正flow1，从消息头获取sourceSystem #

@Bean
public IntegrationFlow flow1() {
  return integrationFlowDefinition ->
      integrationFlowDefinition
          .channel(c -> c.executor(Executors.newCachedThreadPool()))
          .transform(Message.class, message -> {
            try {
              LionRequest payload = (LionRequest) message.getPayload();
              // 直接从透传的请求头中获取枚举值
              SourceSystem sourceSystem = message.getHeaders().get("sourceSystem", SourceSystem.class);
              // dbId挪到这里按请求生成，不要用全局固定值
              long dbId = new SequenceGenerator().nextId();
              return lionService.saveRequest(
                  payload,
                  String.valueOf(dbId),
                  payload.getLionDetails().getId(),
                  sourceSystem.getSourceSystemCode());
            } catch (JsonProcessingException e) {
              return e.getMessage();
            }
          })
          // 如果flow1是异步落库不需要返回结果给聚合器，保留nullChannel即可
          .nullChannel();
}

2. 修正flow3，替换硬编码的枚举值 #

@Bean
public IntegrationFlow flow3() {
  return integrationFlowDefinition ->
      integrationFlowDefinition
          .channel(c -> c.executor(Executors.newCachedThreadPool()))
          .transform(Message.class, message -> {
            LionRequest payload = (LionRequest) message.getPayload();
            SourceSystem sourceSystem = message.getHeaders().get("sourceSystem", SourceSystem.class);
            return lionService.prepareRequest(payload, sourceSystem);
          })
          .handle(
              Http.outboundGateway(someURL)
                  .httpMethod(HttpMethod.POST)
                  .expectedResponseType(String.class));
}

3. flow2如果后续需要使用sourceSystem，用同样方式获取即可 #

@Bean
public IntegrationFlow flow2() {
  return integrationFlowDefinition ->
      integrationFlowDefinition
          .channel(c -> c.executor(Executors.newCachedThreadPool()))
          .transform(Message.class, message -> {
            LionRequest payload = (LionRequest) message.getPayload();
            SourceSystem sourceSystem = message.getHeaders().get("sourceSystem", SourceSystem.class);
            // 按需使用sourceSystem参数
            return loansService.prepareSomething(payload);
          });
}

你的网关代码不需要修改，现有@Header的写法已经可以正确把参数放入消息头，调用方的传参方式也完全符合要求。

复用优化建议 #

如果后续要在多场景复用这套并行处理逻辑，不需要重复编写流配置：

• 可以将Scatter-Gather的核心逻辑抽为公共流定义，将差异化的参数、处理逻辑作为入参传入
• 对于不同SourceSystem对应的差异化处理，可以在子流中加路由逻辑，根据请求头的sourceSystem值选择对应处理分支，避免为每个枚举值维护一套独立流
• 如果flow1确实不需要参与聚合返回结果，可以给scatterer配置recipientFlow时设置applySequence(false)并忽略该流的回复，避免聚合器不必要的等待

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