8.5. 聚合器

借助在 图 8.5 “聚合器模式” 中显示的 聚合器 模式，您可以将相关消息批处理合并到单个消息中。

图 8.5. 聚合器模式

要控制聚合器的行为，Apache Camel 允许您指定 Enterprise Integration Patterns 中描述的属性，如下所示：

例如，考虑一个可每秒接收 30,000 条消息的库存市场数据系统。如果您的 GUI 工具无法与如此大规模的更新率合作，则您可能希望减慢消息流。只需选择最新的报价并丢弃旧的价格，即可聚合传入的库存报价。（如果您愿意捕获历史信息，可以应用 delta 处理算法。）

图 8.6 “聚合器实施” 展示了聚合器如何工作的概览，假设它通过带有关联键的交换流（如 A、B、C 或 D）进行交换流。

图 8.6. 聚合器实施

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

以下示例使用 UseLatestAggregationStrategy 聚合策略来聚合具有相同 VDDK Symbol 标头值的交换。对于给定的 prepare Symbol 值，如果收到了与该关联密钥的最后三秒钟以上，则聚合的交换被视为完成状态并发送到 模拟 端点。

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 命令。这里没有限制使用 Simple 表达式语言。您可以使用任何表达式语言或脚本语言（如 XPath、XQuery、SQL 等）指定关联表达式。

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

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

如果无法在特定的交换中评估关联表达式，聚合器默认会抛出 CamelExchangeException。您可以通过设置 ignoreCorrelationKeys 选项来限制这个异常。例如，在 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 消息时终止库存引号流（如最新传入的交换中 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"/>

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

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 类，可以通过一种机制来强制完成当前消息组，方法是将 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 语言表达式用作使用流模式的 tokenizeXML 子命令的令牌。使用简单语言表达式将支持动态令牌。

例如，要使用 Java 将一系列名称分割成标签用户角色，您可以使用 令牌izeXML 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");

发送至 模拟：result 的分组交换列表包含消息正文中的聚合交换列表。以下行显示后续的处理器如何以列表的形式访问分组交换的内容：

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

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

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

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

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

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

<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>

repositories Name、transactionManager 和 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 实例（在这种情况下，不需要启用 parallelProcessing 选项）。

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

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

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);
    }
}

聚合器支持以下选项：