拆分后并行丰富

Question

这是shopping cart sample 的延续，我们有一个外部 API 允许从购物车中结帐。回顾一下，我们有一个流程，我们创建一个空购物，添加订单项，最后结帐。上述所有操作都是通过对外部服务的 HTTP 调用进行的扩充。我们想同时添加订单项（作为添加订单项的一部分）调用。我们当前的配置如下所示：

@Bean
public IntegrationFlow fullCheckoutFlow() {
    return f -> f.channel("inputChannel")
            .transform(fromJson(ShoppingCart.class))
            .enrich(e -> e.requestChannel(SHOPPING_CART_CHANNEL))
            .split(ShoppingCart.class, ShoppingCart::getLineItems)
            .enrich(e -> e.requestChannel(ADD_LINE_ITEM_CHANNEL))
            .aggregate(aggregator -> aggregator
                    .outputProcessor(g -> g.getMessages()
                            .stream()
                            .map(m -> (LineItem) m.getPayload())
                            .map(LineItem::getName)
                            .collect(joining(", "))))
            .enrich(e -> e.requestChannel(CHECKOUT_CHANNEL))
            .<String>handle((p, h) -> Message.called("We have " + p + " line items!!"));
}

    @Bean
    public IntegrationFlow addLineItem(Executor executor) {
        return f -> f.channel(MessageChannels.executor(ADD_LINE_ITEM_CHANNEL, executor).get())
                .handle(outboundGateway("http://localhost:8080/api/add-line-item", restTemplate())
                        .httpMethod(POST)
                        .expectedResponseType(String.class));
    }

    @Bean
    public Executor executor(Tracer tracer, TraceKeys traceKeys, SpanNamer spanNamer) {
        return new TraceableExecutorService(newFixedThreadPool(10), tracer, traceKeys, spanNamer);
    }

为了并行添加订单项，我们使用了一个执行器通道。但是，当在 zipkin 中看到它们时，它们似乎仍然是按顺序处理的：

我们做错了什么？整个项目的源码在github，供参考。

谢谢！

Answer 1

首先 Spring Integration 的主要特性是 MessageChannel，但我仍然不清楚为什么人们在端点定义之间缺少 .channel() 运算符。

我的意思是，对于你的情况，它必须是这样的：

.split(ShoppingCart.class, ShoppingCart::getLineItems)
.channel(c -> c.executor(executor()))
.enrich(e -> e.requestChannel(ADD_LINE_ITEM_CHANNEL))

现在谈谈你的具体问题。

看，ContentEnricher (.enrich()) 是请求-回复组件：http://docs.spring.io/spring-integration/reference/html/messaging-transformation-chapter.html#payload-enricher。

因此，它向其requestChannel 发送请求并等待回复。而且它是独立于requestChannel 类型完成的。

我是原始Java，我们可以用这段代码sn-p演示这样的行为：

for (Object item: items) {
    Data data = sendAndReceive(item);
}

您应该在哪里看到 ADD_LINE_ITEM_CHANNEL 作为 ExecutorChannel 没有太大价值，因为无论如何我们都被阻止在循环中进行回复。

.split() 执行完全相同的循环，但由于默认情况下它使用DirectChannel，因此迭代在同一个线程中完成。因此，每个下一个项目都等待前一个项目的回复。

这就是为什么您绝对应该在.enrich() 的输入之后完全并行，紧跟在.split() 之后。