java - Spring集成路由到不同类中的子流

Question

为了完成这项工作，我努力了几天。我想要完成的是根据消息内容从主流调用不同的子流(即集成流)，并在子流完成后返回到主流。它就像将责任委托(delegate)给特定的类来完成某些事情并返回到主流。该责任也可能需要一些步骤，因此它也作为流程实现。这是我的主要流程:

public IntegrationFlow processingFlow(
  MessageChannel eventIn,
  MessageChannel eventOut,
  ChangedEventsLoader changedEventsLoader,
  CalculatorRouter calculatorRouter) {

return IntegrationFlows.from(eventIn)
    .handle(changedEventsLoader)
    .route(
        CalculatorRouter::getSportId,
        CalculatorRouter::routeCalculation)
    .channel(eventOut)
    .get();

这里是路由器的实现:

@Service
@AllArgsConstructor
public class CalculatorRouter {
  private final MessageChannel eventOut;

  public RouterSpec<Integer, MethodInvokingRouter> routeCalculation(
      RouterSpec<Integer, MethodInvokingRouter> mapping) {
    return mapping
        .channelMapping(1, "subflowCalculationChannel")
        .defaultOutputToParentFlow();
  }

  public Integer getSportId(Event event) {
    return 1;
  }

  @Bean
  public MessageChannel subflowCalculationChannel() {
    return MessageChannels.direct().get();
  }
}

这是一个子流程的示例:

@Configuration
@AllArgsConstructor
public class CalculatorExample {

  @Bean
  public IntegrationFlow calculateProbabilities(MessageChannel subflowCalculationChannel) {
    return IntegrationFlows.from(subflowCalculationChannel)
        .<Event>handle((p, m) -> p * 2)
        .get();
  }
}

问题是子流丢失了一些与主流的联系。我试图通过在路由部分使用 defaultOutputToParentFlow() 来解决这个问题，但这还不够。

Answer 1

从某个版本开始，我们决定将 Java DSL 路由器行为与带有注释或 XML 的标准配置保持一致。所以，如果我们发送到路由器，我们不能期望从那里得到回复。我们只能继续将 channel 作为子流的输出。

在您的情况下，您在主流中有一个 .channel(eventOut)。所以，你所有的路由子流都应该准确地回复这个 channel :

    .<Event>handle((p, m) -> corners1H2HCustomBet.getCalculation(p))
    .channel(eventOut)
    .get();

我认为 .defaultOutputToParentFlow(); 不会为您做任何事情，因为您没有默认映射。它已经略有不同:它对其他映射没有任何影响。

还要注意这个JavaDoc:

/**
 * Add a subflow as an alternative to a {@link #channelMapping(Object, String)}.
 * {@link #prefix(String)} and {@link #suffix(String)} cannot be used when subflow
 * mappings are used.
 * <p> If subflow should refer to the external {@link IntegrationFlow} bean and
 * there is a requirement to expect reply from there, such a reference should be
 * wrapped with a {@code .gateway()}:
 * <pre class="code">
 * {@code
 *     .subFlowMapping(false, sf -> sf.gateway(evenFlow())))
 * }
 * </pre>
 * @param key the key.
 * @param subFlow the subFlow.
 * @return the router spec.
 */
public RouterSpec<K, R> subFlowMapping(K key, IntegrationFlow subFlow) {

与您基于 channel 的路由配置无关，但将来可能会有用。

更新

这里是 subFlowMapping 的示例 (Kotlin) 并返回主流程:

    @Bean
    fun splitRouteAggregate() =
            IntegrationFlow { f ->
                f.split()
                        .route<Int, Boolean>({ o -> o % 2 == 0 },
                                { m ->
                                    m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
                                            .subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
                                })
                        .aggregate()
            }

    @Bean
    fun oddFlow() =
            IntegrationFlow { flow ->
                flow.handle<Any> { _, _ -> "odd" }
            }

    @Bean
    fun evenFlow() =
            IntegrationFlow { flow ->
                flow.handle<Any> { _, _ -> "even" }
            }