8.3. 接收者列表

接收者列表 （如 图 8.3 “接收者列表模式” ）是将每个传入消息发送到多个不同目的地的路由器类型。另外，接收者列表通常要求在运行时计算接收者列表。

图 8.3. 接收者列表模式

最简单的接收者列表是目标列表已修复且提前已知的，交换模式是 InOnly。在这种情况下，您可以将目的地列表分成 to （） Java DSL 命令。

以下示例演示了如何将来自消费者端点 队列：a 的 InOnly 交换路由到目的地的固定列表：

from("seda:a").to("seda:b", "seda:c", "seda:d");

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

<camelContext id="buildStaticRecipientList" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="seda:a"/>
    <to uri="seda:b"/>
    <to uri="seda:c"/>
    <to uri="seda:d"/>
  </route>
</camelContext>

在大多数情况下，当使用接收者列表模式时，应在运行时计算接收方列表。为此，可使用 接收者List （） 处理器，它取目的地列表作为其唯一参数。因为 Apache Camel 将类型转换器应用于 list 参数，因此应该可以使用大多数标准 Java 列表类型（如集合、列表或数组）。有关类型转换器的详情，请参考 第 34.3 节 “内置类型 Converters”。

接收者收到 同一 交换实例的副本，并且 Apache Camel 按顺序执行它们。

以下示例演示了如何从名为 recipientListHeader 的消息标头中提取目的地列表，其中标头值是一个以逗号分隔的端点 URI 列表：

from("direct:a").recipientList(header("recipientListHeader").tokenize(","));

在某些情况下，如果标头值是一个列表类型，您可以直接使用它作为 接收者List （） 的参数。例如：

from("seda:a").recipientList(header("recipientListHeader"));

但是，本例完全依赖于底层组件如何解析这个特定标头。如果组件以简单字符串的形式解析标头，本例 将无法工作。标头必须解析到某些类型的 Java 列表中。

以下示例演示了如何在 XML 中配置上述路由，其中 标头值是一个以逗号分隔的端点 URI 列表：

<camelContext id="buildDynamicRecipientList" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="seda:a"/>
    <recipientList delimiter=",">
      <header>recipientListHeader</header>
    </recipientList>
  </route>
</camelContext>

可作为 Camel 2.2 提供

接收者列表模式 支持 并行处理，这与 拆分器模式 中的对应 功能类似。使用并行处理功能将交换发送到同时到达多个接收方，例如：

from("direct:a").recipientList(header("myHeader")).parallelProcessing();

在 Spring XML 中，并行处理功能是以 接收者List tag 的 attribute 进行实现的，例如：

<route>
  <from uri="direct:a"/>
  <recipientList parallelProcessing="true">
    <header>myHeader</header>
  </recipientList>
</route>

可作为 Camel 2.2 提供

接收者列表 支持 stopOnException 功能，您可以在任何接收者失败时停止发送到任何进一步收件人。

from("direct:a").recipientList(header("myHeader")).stopOnException();

在 Spring XML 中，它在接收者列表标签中有一个属性。

在 Spring XML 中，停止异常功能是以 接收者List tag 的 attribute 进行实现的，例如：

<route>
  <from uri="direct:a"/>
  <recipientList stopOnException="true">
    <header>myHeader</header>
  </recipientList>
</route>

可作为 Camel 2.3 提供

接收者列表模式 支持 ignoreInvalidEndpoints 选项，该选项可让接收者列表跳过无效的端点(路由 slips 模式 也支持这个选项)。例如：

from("direct:a").recipientList(header("myHeader")).ignoreInvalidEndpoints();

在 Spring XML 中，您可以通过在 recipientList 标签上设置 ignoreInvalidEndpoints 属性来启用这个选项，如下所示

<route>
  <from uri="direct:a"/>
  <recipientList ignoreInvalidEndpoints="true">
    <header>myHeader</header>
  </recipientList>
</route>

考虑 myHeader 包含两个端点( direct:foo,xxx:bar )的大小写。第一个端点有效并可工作。第二个操作无效，因此忽略。每当遇到无效的端点时，会位于 INFO 级别的 Apache Camel 日志。

可作为 Camel 2.2 提供

您可以将自定义的 AggregationStrategy 与 接收者列表模式 搭配使用，这对于聚合来自列表中的接收方的回复非常有用。默认情况下，Apache Camel 使用 UseLatestAggregationStrategy 聚合策略，该策略只保留上次收到的回复。对于更复杂的聚合策略，您可以自行定义 AggregationStrategy interfaceTOKEN-确保 第 8.5 节 “聚合器” 的实现。详情请参阅。例如，要将自定义聚合策略 MyOwnAggregationStrategy 应用到回复消息，您可以按照如下所示定义 Java DSL 路由：

from("direct:a")
    .recipientList(header("myHeader")).aggregationStrategy(new MyOwnAggregationStrategy())
    .to("direct:b");

在 Spring XML 中，您可以将自定义聚合策略指定为 recipientList 标签上的属性，如下所示：

<route>
  <from uri="direct:a"/>
  <recipientList strategyRef="myStrategy">
    <header>myHeader</header>
  </recipientList>
  <to uri="direct:b"/>
</route>

<bean id="myStrategy" class="com.mycompany.MyOwnAggregationStrategy"/>

可作为 Camel 2.2 提供

这只在您使用 并行处理时才需要。默认情况下，Camel 使用了 10 个线程的线程池。请注意，当我们彻底掌握的线程池管理和以后配置时，这可能会改变（在 Camel 2.2 中）。

就像使用自定义聚合策略一样配置它。

您可以使用 bean 集成来提供接收者，例如：

from("activemq:queue:test").recipientList().method(MessageRouter.class, "routeTo");

MessageRouter bean 的定义位置：

public class MessageRouter {

    public String routeTo() {
        String queueName = "activemq:queue:test2";
        return queueName;
    }
}

您可以通过将 @RecipientList 注释添加到返回接收者列表的方法，使 bean 作为接收者列表的行为。例如：

public class MessageRouter {

    @RecipientList
    public String routeTo() {
        String queueList = "activemq:queue:test1,activemq:queue:test2";
        return queueList;
    }
}

在本例中，不要在 路由中包含 接收者List DSL 命令。按照如下所示定义路由：

from("activemq:queue:test").bean(MessageRouter.class, "routeTo");

可作为 Camel 2.5 提供

如果使用 parallelProcessing，可以用毫秒来配置一个总 超时值。Camel 随后会并行处理消息，直到超时达到为止。如果一条消息慢，则可以继续处理。

在以下示例中，接收者列表标头具有值 direct:a,direct:b,direct:c，因此消息会发送至三个收件人。我们有 250 毫秒的超时，这意味着在时间段内只能完成最后两个消息。因此，聚合会产生字符串 BC。

from("direct:start")
    .recipientList(header("recipients"), ",")
    .aggregationStrategy(new AggregationStrategy() {
            public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                if (oldExchange == null) {
                    return newExchange;
                }

                String body = oldExchange.getIn().getBody(String.class);
                oldExchange.getIn().setBody(body + newExchange.getIn().getBody(String.class));
                return oldExchange;
            }
        })
        .parallelProcessing().timeout(250)
    // use end to indicate end of recipientList clause
    .end()
    .to("mock:result");

from("direct:a").delay(500).to("mock:A").setBody(constant("A"));

from("direct:b").to("mock:B").setBody(constant("B"));

from("direct:c").to("mock:C").setBody(constant("C"));

默认情况下，如果没有调用 AggregationStrategy，则默认会发生超时。但是，您可以实现一个特殊的版本

// Java
public interface TimeoutAwareAggregationStrategy extends AggregationStrategy {

    /**
     * A timeout occurred
     *
     * @param oldExchange  the oldest exchange (is <tt>null</tt> on first aggregation as we only have the new exchange)
     * @param index        the index
     * @param total        the total
     * @param timeout      the timeout value in millis
     */
    void timeout(Exchange oldExchange, int index, int total, long timeout);

如果您真正需要，这可以处理 AggregationStrategy 中的超时时间。

在 接收者List 将消息发送到接收者端点之一之前，它会创建一个消息副本，即原始消息的应复制。在匹配项中，原始消息的标头和有效负载仅通过引用来复制。每个新副本不包含这些元素自己的实例。因此，当消息被链接，您应该在将自定义处理路由到不同的端点时，您无法应用自定义处理。

如果要在将副本发送到端点之前对每条消息副本执行一些自定义处理，您可以在 接收者List 子句中调用 onPrepare DSL 命令。onPrepare 命令仅在消息 被禁止后 插入自定义处理器，且仅在消息被分配到其端点 之前。例如，在以下路由中，会在每个接收者端点 的消息副本上调用 CustomProc 处理器：

from("direct:start")
  .recipientList().onPrepare(new CustomProc());

onPrepare DSL 命令的一个常见用例是执行对消息的某些或所有元素的深度副本。这允许独立于其他消息副本修改每个消息副本。例如，以下 CustomProc 处理器类对消息正文执行深度副本，其中消息正文假定为 type、BrodyType，并且深度副本由方法 BodyType .deepCopy （） 执行。

// Java
import org.apache.camel.*;
...
public class CustomProc implements Processor {

    public void process(Exchange exchange) throws Exception {
        BodyType body = exchange.getIn().getBody(BodyType.class);

        // Make a _deep_ copy of of the body object
        BodyType clone =  BodyType.deepCopy();
        exchange.getIn().setBody(clone);

        // Headers and attachments have already been
        // shallow-copied. If you need deep copies,
        // add some more code here.
    }
}

接收者List DSL 命令支持以下选项：

默认情况下，Recipient List 使用当前的交换模式。但是，在有些情况下，您可以使用不同的交换模式向接收者发送信息。

例如，您可能有一个作为 InOnly 路由启动的路由。现在，如果要将 InOut Exchange 模式与接收者列表搭配使用，则需要直接在接收者端点中配置交换模式。

以下示例说明了新文件将作为 InOnly 启动的路由，然后路由到接收者列表。如果您想将 InOut 与 ActiveMQ (JMS)端点搭配使用，则需要使用与 InOut 选项的 exchangePattern 来指定它。但是，对 JMS 请求或回复形成的响应将被持续路由，因此响应作为 outbox 目录中的文件存储在 中。

from("file:inbox")
  // the exchange pattern is InOnly initially when using a file route
  .recipientList().constant("activemq:queue:inbox?exchangePattern=InOut")
  .to("file:outbox");