8.5. 聚合器

通过 图 8.5 “聚合器模式” 所示的 聚合器 模式，您可以将相关消息批量合并到一个消息中。

图 8.5. 聚合器模式

要控制聚合器的行为，Apache Camel 允许您指定 企业集成模式 中描述的属性，如下所示：

例如，一个库存市场数据系统，每秒接收 30,000 个消息。如果您的 GUI 工具无法应对此类更新率，您可能希望降低消息流。只需选择最新的报价并丢弃旧价格，即可将传入股票报价聚合在一起。（如果您愿意捕获一些历史记录，可以应用 delta 处理算法。）

图 8.6 “聚合器实施” 显示聚合器如何工作的概述，假设它带有与 A、B、C 或 D 等关联键（如 A、B、C 或 D）的交换流。

图 8.6. 聚合器实施

图 8.6 “聚合器实施” 中显示的交换流处理如下：

以下示例使用 UseLatestAggregationStrategy 聚合策略使用同一 StockSymbol 标头值来聚合交换。对于给定的 StockSymbol 值，如果自收到了最后的关联密钥交换了三秒以上，则聚合交换被视为完成并发送到 模拟 端点。

from("direct:start")
    .aggregate(header("id"), new UseLatestAggregationStrategy())
        .completionTimeout(3000)
    .to("mock:aggregated");

以下示例演示了如何在 XML 中配置相同的路由：

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <aggregate strategyRef="aggregatorStrategy"
                   completionTimeout="3000">
            <correlationExpression>
                <simple>header.StockSymbol</simple>
            </correlationExpression>
            <to uri="mock:aggregated"/>
        </aggregate>
    </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.aggregate.UseLatestAggregationStrategy"/>

在 Java DSL 中，关联表达式始终作为参数传递到 aggregate （） DSL 命令。您不限于此处使用简单表达式语言。您可以使用任何表达式语言或脚本语言（如 XPath、XQuery、SQL 等）指定关联表达式。

对于 exampe，要使用 XPath 表达式来关联交换，您可以使用以下 Java DSL 路由：

from("direct:start")
    .aggregate(xpath("/stockQuote/@symbol"), new UseLatestAggregationStrategy())
        .completionTimeout(3000)
    .to("mock:aggregated");

如果无法对特定传入的交换进行评估，则聚合器默认会引发 CamelExchangeException。您可以通过设置 ignoreInvalidCorrelationKeys 选项来抑制此异常。例如，在 Java DSL 中：

from(...).aggregate(...).ignoreInvalidCorrelationKeys()

在 XML DSL 中，您可以设置 ignoreInvalidCorrelationKeys 选项作为属性，如下所示：

<aggregate strategyRef="aggregatorStrategy"
           ignoreInvalidCorrelationKeys="true"
           ...>
    ...
</aggregate>

在 Java DSL 中，您可以将聚合策略作为第二参数传递给 aggregate （） DSL 命令，或使用 aggregationStrategy （） 子句指定它。例如，您可以按照如下所示使用 aggregationStrategy （） 子句：

from("direct:start")
    .aggregate(header("id"))
        .aggregationStrategy(new UseLatestAggregationStrategy())
        .completionTimeout(3000)
    .to("mock:aggregated");

Apache Camel 提供以下基本的聚合策略（其中类属于 org.apache.camel.processor.aggregate Java 软件包）：

如果要应用不同的聚合策略，您可以实现以下聚合策略基本接口之一：

例如，以下代码显示了两个不同的自定义聚合策略，StringAggregationStrategy 和 ArrayListAggregationStrategy:

 //simply combines Exchange String body values using '' as a delimiter
 class StringAggregationStrategy implements AggregationStrategy {

     public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
         if (oldExchange == null) {
             return newExchange;
         }

         String oldBody = oldExchange.getIn().getBody(String.class);
         String newBody = newExchange.getIn().getBody(String.class);
         oldExchange.getIn().setBody(oldBody + "" + newBody);
         return oldExchange;
     }
 }

 //simply combines Exchange body values into an ArrayList<Object>
 class ArrayListAggregationStrategy implements AggregationStrategy {

     public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
 	    Object newBody = newExchange.getIn().getBody();
     	ArrayList<Object> list = null;
         if (oldExchange == null) {
 		    list = new ArrayList<Object>();
 		    list.add(newBody);
 		    newExchange.getIn().setBody(list);
 		    return newExchange;
         } else {
 	        list = oldExchange.getIn().getBody(ArrayList.class);
 	    	list.add(newBody);
 		    return oldExchange;
 	    }
     }
 }

要使用自定义策略类 ArrayListAggregationStrategy 来聚合信息，请定义类似以下的路由：

from("direct:start")
    .aggregate(header("StockSymbol"), new ArrayListAggregationStrategy())
    .completionTimeout(3000)
    .to("mock:result");

您还可以使用 XML 中的自定义聚合策略配置路由，如下所示：

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start"/>
    <aggregate strategyRef="aggregatorStrategy"
               completionTimeout="3000">
      <correlationExpression>
        <simple>header.StockSymbol</simple>
      </correlationExpression>
      <to uri="mock:aggregated"/>
    </aggregate>
  </route>
</camelContext>

<bean id="aggregatorStrategy" class="com.my_package_name.ArrayListAggregationStrategy"/>

您可以实施自定义聚合策略，以便其生命周期与控制它的企业集成模式的生命周期一致。这有助于确保聚合策略可以正常关闭。

要实施具有生命周期支持的聚合策略，您必须实施 org.apache.camel.Service 接口（在 AggregationStrategy 接口之外），并提供 start （） 和 stop （） 生命周期方法的实施。例如，以下代码示例显示了一个带有生命周期支持的聚合策略概述：

// Java
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.apache.camel.Service;
import java.lang.Exception;
...
class MyAggStrategyWithLifecycleControl
       implements AggregationStrategy, Service {

    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        // Implementation not shown...
        ...
    }

    public void start() throws Exception {
        // Actions to perform when the enclosing EIP starts up
        ...
    }

    public void stop() throws Exception {
        // Actions to perform when the enclosing EIP is stopping
        ...
    }
}

每个聚合交换中都会设置以下属性：

以下属性是在被 SQL 组件聚合仓库重新提供的交换上设置（请参阅 “持久聚合存储库”一节）：

至少需要指定一个 完成条件，决定聚合交换何时离开聚合器并继续进行路由上的下一个节点。可以指定以下完成条件：

前面的完成条件可以任意组合，但 completionTimeout 和 completionInterval 条件除外，这些条件无法同时启用。当条件结合使用时，常规规则是触发的第一个完成条件是有效的完成条件。

您可以指定任意 predicate 表达式，决定何时完成聚合交换。评估 predicate 表达式的方法有两种：

例如，如果您想要在每次收到 ALERT 消息时终止 stock quotes 流（根据最新传入交换中的 MsgType 标头的值所示），您可以定义类似如下的路由：

from("direct:start")
    .aggregate(
      header("id"),
      new UseLatestAggregationStrategy()
    )
        .completionPredicate(
          header("MsgType").isEqualTo("ALERT")
         )
        .eagerCheckCompletion()
    .to("mock:result");

以下示例演示了如何使用 XML 配置相同的路由：

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start"/>
    <aggregate strategyRef="aggregatorStrategy"
               eagerCheckCompletion="true">
      <correlationExpression>
          <simple>header.StockSymbol</simple>
      </correlationExpression>
      <completionPredicate>
          <simple>$MsgType = 'ALERT'</simple>
      </completionPredicate>
      <to uri="mock:result"/>
    </aggregate>
  </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.aggregate.UseLatestAggregationStrategy"/>

可以指定 动态完成超时，其中每个传入交换都会重新计算超时值。例如，若要从每个传入交换的 timeout 标头设置超时值，您可以按照如下所示定义路由：

from("direct:start")
    .aggregate(header("StockSymbol"), new UseLatestAggregationStrategy())
        .completionTimeout(header("timeout"))
    .to("mock:aggregated");

您可以在 XML DSL 中配置相同的路由，如下所示：

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <aggregate strategyRef="aggregatorStrategy">
            <correlationExpression>
                <simple>header.StockSymbol</simple>
            </correlationExpression>
            <completionTimeout>
                <header>timeout</header>
            </completionTimeout>
            <to uri="mock:aggregated"/>
        </aggregate>
    </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.UseLatestAggregationStrategy"/>

可以指定 动态完成大小，其中为每个传入交换重新计算完成大小。例如，若要从每个传入交换的 mySize 标头设置完成大小，您可以按照如下所示定义路由：

from("direct:start")
    .aggregate(header("StockSymbol"), new UseLatestAggregationStrategy())
        .completionSize(header("mySize"))
    .to("mock:aggregated");

使用 Spring XML 的相同示例：

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <aggregate strategyRef="aggregatorStrategy">
            <correlationExpression>
                <simple>header.StockSymbol</simple>
            </correlationExpression>
            <completionSize>
                <header>mySize</header>
            </completionSize>
            <to uri="mock:aggregated"/>
        </aggregate>
    </route>
</camelContext>

<bean id="aggregatorStrategy"
      class="org.apache.camel.processor.UseLatestAggregationStrategy"/>

如果您实施了自定义 AggregationStrategy 类，有一个机制来强制完成当前消息组，方法是在从 AggregationStrategy.aggregate （） 方法返回的交换时将 Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP Exchange 属性设置为 true。这个机制 仅影响 当前的组：其他消息组（具有不同的关联 ID） 不会被 强制完成。这种机制会覆盖任何其他完成机制，如 predicate、大小、超时等。

例如，如果消息正文大小大于 5，则以下示例 AggregationStrategy 类将完成当前的组：

// Java
public final class MyCompletionStrategy implements AggregationStrategy {
    @Override
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        if (oldExchange == null) {
            return newExchange;
        }
        String body = oldExchange.getIn().getBody(String.class) + "+"
            + newExchange.getIn().getBody(String.class);
        oldExchange.getIn().setBody(body);
        if (body.length() >= 5) {
            oldExchange.setProperty(Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP, true);
        }
        return oldExchange;
    }
}

通过向路由发送带有特殊标头的消息，可以强制完成所有未完成的聚合消息。您可以使用两个替代的标头设置来强制完成：

org.apache.camel.processor.aggregate.AggregateController 可让您使用 Java 或 JMX API 在运行时控制聚合。这可用于强制完成一组交换，或者查询当前的运行时统计数据。

如果没有配置自定义，则聚合器提供了一个默认实现，您可以使用 getAggregateController （） 方法访问。但是，使用 aggregateController 在路由中配置控制器很容易。

private AggregateController controller = new DefaultAggregateController();

from("direct:start")
   .aggregate(header("id"), new MyAggregationStrategy()).completionSize(10).id("myAggregator")
      .aggregateController(controller)
      .to("mock:aggregated");

另外，您可以使用 AggregateController上的 API 来强制完成。例如，使用密钥 foo 完成组

int groups = controller.forceCompletionOfGroup("foo");

数字返回可以是已完成的组数。以下是一个 API 以完成所有组：

 int groups = controller.forceCompletionOfAllGroups();

在一些聚合场景中，您可能需要强制关联密钥对于每批交换是唯一的条件。换句话说，当特定关联键的聚合交换完成时，您希望确保不允许与该关联密钥进一步的聚合交换。例如，如果路由的后面的部分应该与唯一关联键值交换，您可能希望强制实施此条件。

根据配置完成条件的方式，可能会有多个使用特定关联密钥生成的聚合交换的风险。例如，虽然您可能会定义一个完成 predicate，它设计为等待所有与特定相关性密钥交换的交换被接收，但您可能还会定义一个完成超时，这可以在与该密钥的所有交换到达之前触发。在这种情况下，相关的交换可能会增加 第二个 聚合交换，具有相同相关性的键值。

对于这样的情形，您可以通过设置 closeCorrelationKeyOnCompletion 选项，将聚合器配置为禁止聚合与之前关联密钥值的聚合交换。为了禁止重复的关联键值，聚合器需要记录之前的关联键值。此缓存的大小（缓存的关联键的数量）指定为 closeCorrelationKeyOnCompletion （） DSL 命令的参数。要指定无限大小的缓存，您可以传递值零或负整数。例如，要指定 10000 个键值的缓存大小：

from("direct:start")
    .aggregate(header("UniqueBatchID"), new MyConcatenateStrategy())
        .completionSize(header("mySize"))
        .closeCorrelationKeyOnCompletion(10000)
    .to("mock:aggregated");

如果聚合交换使用重复的关联键值完成，则聚合器会抛出一个 ClosedCorrelationKeyException 异常。

您可以使用 Simple 语言表达式作为令牌，在 streaming 模式中使用 tokenizeXML 子命令。使用简单语言表达式将启用对动态令牌的支持。

例如，要使用 Java 将一系列按标签用户角色分离的名称分隔，您可以使用 令牌化XML bean 和简单语言令牌 将该文件拆分为 名称 元素。

public void testTokenizeXMLPairSimple() throws Exception {
        Expression exp = TokenizeLanguage.tokenizeXML("${header.foo}", null);

获取由 < person> 划分的名称的 输入字符串，并将 &lt ;person& gt; 设置为令牌。

        exchange.getIn().setHeader("foo", "<person>");
        exchange.getIn().setBody("<persons><person>James</person><person>Claus</person><person>Jonathan</person><person>Hadrian</person></persons>");

列出从输入中分离的名称。

        List<?> names = exp.evaluate(exchange, List.class);
        assertEquals(4, names.size());

        assertEquals("<person>James</person>", names.get(0));
        assertEquals("<person>Claus</person>", names.get(1));
        assertEquals("<person>Jonathan</person>", names.get(2));
        assertEquals("<person>Hadrian</person>", names.get(3));
    }

您可以将传出批处理中的所有聚合交换合并为一个 org.apache.camel.impl.GroupedExchange holder 类。要启用分组的交换，请指定 groupExchanges （） 选项，如以下 Java DSL 路由中所示：

from("direct:start")
    .aggregate(header("StockSymbol"))
        .completionTimeout(3000)
        .groupExchanges()
    .to("mock:result");

发送到 mock:result 的分组交换列表包含消息正文中聚合交换的列表。以下代码行显示了后续处理器如何以列表的形式访问所分组交换的内容：

// Java
List<Exchange> grouped = ex.getIn().getBody(List.class);

聚合器可以与 批处理消费者模式一同工作，以汇总批处理消费者 报告的消息总数（批处理消费者端点设置 CamelBatchSize、CamelBatchIndex、CamelBatchIndex、CamelBatchIndex 的属性）和 CamelBatchComplete 属性。例如，要聚合由文件消费者端点找到的所有文件，您可以使用类似如下的路由：

from("file://inbox")
    .aggregate(xpath("//order/@customerId"), new AggregateCustomerOrderStrategy())
    .completionFromBatchConsumer()
    .to("bean:processOrder");

目前，以下端点支持批处理使用者机制：file、FTP、邮件、iBatis 和 JPA。

默认聚合器仅使用内存中的 AggregationRepository。如果要永久存储待处理的聚合交换，您可以使用 SQL 组件 作为持久聚合存储库。SQL 组件包含一个 JdbcAggregationRepository，它可即时保留聚合的信息，并确保您不会丢失任何信息。

成功处理交换时，当存储库上调用 确认方法时，它将标记为完成。这意味着，如果同一交换再次失败，它将重试，直到成功为止。

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-sql</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

CREATE TABLE my_aggregation_repo (
 id varchar(255) NOT NULL,
 exchange blob NOT NULL,
 constraint aggregation_pk PRIMARY KEY (id)
);

CREATE TABLE my_aggregation_repo_completed (
 id varchar(255) NOT NULL,
 exchange blob NOT NULL,
 constraint aggregation_completed_pk PRIMARY KEY (id)
);
}

<bean id="my_repo"
    class="org.apache.camel.processor.aggregate.jdbc.JdbcAggregationRepository">
    <property name="repositoryName" value="my_aggregation_repo"/>
    <property name="transactionManager" ref="my_tx_manager"/>
    <property name="dataSource" ref="my_data_source"/>
    ...
</bean>

repositoryName、事务管理器 和 dataSource 属性是必需的。有关持久聚合存储库配置选项的详情，请参阅 Apache Camel 组件参考指南中的 SQL 组件。

如 图 8.6 “聚合器实施” 所示，聚合器与路由的后一部分分离，该交换发送到路由的后部分，由专用线程池处理。默认情况下，这个池仅包含一个线程。如果要指定有多个线程的池，启用 parallelProcessing 选项，如下所示：

from("direct:start")
    .aggregate(header("id"), new UseLatestAggregationStrategy())
        .completionTimeout(3000)
        .parallelProcessing()
    .to("mock:aggregated");

默认情况下，这会创建一个具有 10 个 worker 线程的池。

如果要对创建的线程池进行更多控制，请使用 executorService 选项指定自定义 java.util.concurrent.ExecutorService 实例（在这种情况下，不需要启用 并行处理 选项）。

常见的聚合方案涉及将一系列传入的消息正文整合到 List 对象中。为便于这种情况，Apache Camel 提供了 AbstractListAggregationStrategy 抽象类，您可以快速扩展以针对此案例创建聚合策略。传入类型消息 T 聚合到完成的交换中，消息正文类型为 List<T>。

例如，要将一系列 Integer 消息正文聚合到一个 List<Integer& gt; 对象中，您可以使用如下定义的聚合策略：

import org.apache.camel.processor.aggregate.AbstractListAggregationStrategy;
...
/**
 * Strategy to aggregate integers into a List<Integer>.
 */
public final class MyListOfNumbersStrategy extends AbstractListAggregationStrategy<Integer> {
 
    @Override
    public Integer getValue(Exchange exchange) {
        // the message body contains a number, so just return that as-is
        return exchange.getIn().getBody(Integer.class);
    }
}

聚合器支持以下选项：