java - Apache Camel 聚合器 : Input and Output

Question

我有以下路线:

<route id="my-aggregator">
    <from uri="direct:aggregator" />

    <aggregate strategyRef="myAggregationStrategy" completionSize="3">
        <correlationExpression>
            <simple>${header.id}</simple>
        </correlationExpression>
        <to uri="bean:payloadProcessor?method=process" />
    </aggregate>
</route>

如您所见，它会一直等待，直到收到 3 个具有匹配 header.id 的交换。值(value)。一旦收到 3 条这样的消息，就会执行以下代码:

public class MyAggregationStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange e1, Exchange e2) {
        // The first (of 3) message will have a `List<Widget>` on its body. Extract it here.
        List<Widget> redWidgets = ??? // e1.getIn().getBody()? e2.getIn().getBody()? something else?

        // The second (of 3) message will have a `List<Widget>` on its body. Extract it here.
        List<Widget> blueWidgets = ??? // e1.getIn().getBody()? e2.getIn().getBody()? something else?

        // The third (of 3) message will have a `List<Widget>` on its body. Extract it here.
        List<Widget> greenWidgets = ??? // e1.getIn().getBody()? e2.getIn().getBody()? something else?

        WidgetPayload payload = new WidgetPayload(redWidgets, blueWidgets, greenWidgets);

        // Which "output" exchange do I set payload to, so that it gets routed on to the
        // "payloadProcessor" bean?
        e1.getOut().setBody(payload);   /* or */    e2.getOut().setBody(payload);

        // What do we even return?
        return e1; /* or return e2; */
    }
}

因此，在聚合器中，我试图访问每个 List<Widget>通过各种路线发送到聚合器，将它们组合成 WidgetPayload实例，然后设置WidgetPayload作为出站交换的出站/返回值。但我无法找出完成这一切的正确方法。具体来说:

1. 如何提取 3 个 List<Widget> 中的每一个来自e1和e2 ？
2. 做什么 e1和e2甚至代表？进出？新旧？还有别的事吗？
3. 什么方法调用 e1/e2我这样做payload转发至 payloadProcessor bean ？

Answer 1

看看我提到的页面的代码示例；第一个参数是聚合交换。第一次调用聚合时它为空。您必须在第一次调用时创建 WidgetPayload，然后在每次后续调用时不断向其中添加小部件。然后，当您到达聚合器之后的路由步骤时，交换的主体就是您聚合的 WidgetPayload。