时间戳

首页 > 解决方案 > 如何将 GroupedExchangeAggregationStrategy 聚合交换拆分为原始交换？

kotlin - 如何将 GroupedExchangeAggregationStrategy 聚合交换拆分为原始交换？

问题描述

在使用 a 聚合交换之后，GroupedExchangeAggregationStrategy我需要将它们分开（以发出单独的处理时间指标）到原始交换中。

我尝试使用以下拆分，但生成的拆分交换包装了原始交换并将其放入Message正文中。

是否可以在GroupedExchangeAggregationStrategy没有包装器交换的情况下将聚合交换拆分为原始交换？我需要使用原始交换属性，并希望使用 SpEL 表达式来实现。

.aggregate(constant(true), myGroupedExchangeAggregationStrategy)
    .completionInterval(1000)
    .completeAllOnStop()
    .process { /* do stuff */ }
.split(exchangeProperty(Exchange.GROUPED_EXCHANGE))
    .to(/* micrometer timer metric using SpEL expression */)
    // ^- the resulting split exchange is wrapped in another exchange

如果当前不支持此功能，我正在尝试找出自行实现此行为的最佳方法，而无需Splitter为此单个功能创建自定义处理器。我希望以某种方式覆盖进行SplitterIterable包装的，但这似乎是不可能的。

标签： kotlinapache-camelspring-camel

解决方案

是的，除了创建一个所有的ExchangesGroupedExchangeAggregationStrategy之外什么也没做。另一方面，Splitter EIP 默认将 List 拆分为元素并将元素放入消息正文中。因此，您最终会得到一个在其主体中包含一个 Exchange 的 Exchange 。java.util.List

您需要的是一个 AggregationStrategy，它收集列表中的所有主体对象，而不是所有交换。

您可以尝试使用可通过 fluent API 配置的CamelsFlexibleAggregationStrategy。

new FlexibleAggregationStrategy() .storeInBody() .accumulateInCollection(ArrayList.class) .pick(new SimpleExpression("${body}"));

这应该创建一个 AggregationStrategy 来提取每条消息的正文（您也许可以省略 pick 方法，因为正文提取是默认选择），将它们收集在 List 中并将聚合存储在消息正文中。

要再次拆分此聚合，一个简单的split(body())就足够了。

编辑由于评论

是的，你是对的，我的解决方案的一个副作用是你丢失了原始消息的属性和标题，因为它只聚合消息正文。

您想要做的是将交换列表拆分回原始列表。即拆分器不能创建新的交换，而是使用已经存在的交换并丢弃聚合的包装交换。

据我在 Splitter 的源代码中可以看到，目前这是不可能的：

Exchange newExchange = ExchangeHelper.createCorrelatedCopy(copy, false);
...
if (part instanceof Message) {
    newExchange.setIn((Message) part);
} else {
    Message in = newExchange.getIn();
    in.setBody(part);
}

推荐阅读