8.4. Splitter

拆分器 是一种路由器类型，可将传入的消息分成一系列传出消息。每个传出消息都包含一组原始消息。在 Apache Camel 中，在 图 8.4 “Splitter Pattern” 中显示的分割模式由 split （） Java DSL 命令实现。

图 8.4. Splitter Pattern

Apache Camel 分割器实际支持两种模式，如下所示：

在将原始消息划分为多个部分之前，它会做原始消息的副本。在 shouldow copy 中，原始消息的标头和有效负载仅复制为参考。虽然 splitter 本身不会将生成的消息部分路由到不同的端点，但分割消息的部分内容可能会进行二级路由。

由于消息部分为 shouldow 副本，所以它们仍然与原始消息相关联。因此，无法独立修改它们。如果要将自定义逻辑应用到消息部分的不同副本，并将其路由到一组端点，则必须使用 splitter 子句中的 onPrepareRef DSL 选项生成原始消息的深度副本。有关使用选项的详情，请参考 “选项”一节。

以下示例定义了从 seda:a 到 seda:b 的路由，该路由通过将传入消息的每一行转换为一个单独的传出消息来分割信息：

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
        from("seda:a")
          .split(bodyAs(String.class).tokenize("\n"))
          .to("seda:b");
    }
};

分割器可以使用任何表达式语言，因此您可以使用任何受支持的脚本语言（如 XPath、XQuery 或 SQL）分割信息（请参阅 第 II 部分 “路由表达式和 predicates 语言”）。以下示例从传入的信息中提取 bar 元素，并将其插入到单独的传出消息中：

from("activemq:my.queue")
  .split(xpath("//foo/bar"))
  .to("file://some/directory")

以下示例演示了如何使用 XPath 脚本语言在 XML 中配置分割路由：

<camelContext id="buildSplitter" xmlns="http://camel.apache.org/schema/spring">
    <route>
      <from uri="seda:a"/>
      <split>
        <xpath>//foo/bar</xpath>
        <to uri="seda:b"/>
      </split>
    </route>
</camelContext>

您可以使用 XML DSL 中的令牌表达式来利用令牌分割正文或标头，其中使用 tokenize 元素定义令牌化表达式。在以下示例中，消息正文使用 \n 分隔符字符进行令牌化。要使用正则表达式模式，请在 tokenize 元素中设置 regex=true。

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="direct:start"/>
        <split>
            <tokenize token="\n"/>
            <to uri="mock:result"/>
        </split>
    </route>
    </camelContext>

要将大型文件分成 1000 行的块，您可以在 Java DSL 中定义分割路由：

from("file:inbox")
    .split().tokenize("\n", 1000).streaming()
       .to("activemq:queue:order");

要令牌化的第二个参数指定应分组到单个块中的行数。streaming （） 子句指示 splitter 不会一次读取整个文件（如果文件较大，则达到更好的性能）。

相同的路由可以在 XML DSL 中定义，如下所示：

<route>
  <from uri="file:inbox"/>
  <split streaming="true">
    <tokenize token="\n" group="1000"/>
    <to uri="activemq:queue:order"/>
  </split>
</route>

使用 group 选项时的输出始终为 java.lang.String 类型。

要跳过消息中的第一个项目，您可以使用 skipFirst 选项。

在 Java DSL 中，在 tokenize 参数 true 中进行第三个选项：

from("direct:start")
 // split by new line and group by 3, and skip the very first element
      .split().tokenize("\n", 3, true).streaming()
         .to("mock:group");

相同的路由可以在 XML DSL 中定义，如下所示：

<route>
  <from uri="file:inbox"/>
    <split streaming="true">
    <tokenize token="\n" group="1000" skipFirst="true" />
    <to uri="activemq:queue:order"/>
  </split>
</route>

如果进入 splitter 的交换具有 InOut message-exchange 模式（即回复是预期的），则拆分器会返回原始输入消息的副本作为 Out 消息插槽中的回复消息。您可以通过实施自己的 聚合策略来覆盖此默认行为。

如果要并行执行生成的消息，您可以启用并行处理选项，该选项实例化一个线程池来处理消息片段。例如：

XPathBuilder xPathBuilder = new XPathBuilder("//foo/bar");
from("activemq:my.queue").split(xPathBuilder).parallelProcessing().to("activemq:my.parts");

您可以自定义并行分割器中使用的底层 ThreadPoolExecutor。例如，您可以在 Java DSL 中指定自定义 executor，如下所示：

XPathBuilder xPathBuilder = new XPathBuilder("//foo/bar");
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());
from("activemq:my.queue")
  .split(xPathBuilder)
  .parallelProcessing()
  .executorService(threadPoolExecutor)
  .to("activemq:my.parts");

您可以在 XML DSL 中指定自定义 executor，如下所示：

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:parallel-custom-pool"/>
    <split executorServiceRef="threadPoolExecutor">
      <xpath>/invoice/lineItems</xpath>
      <to uri="mock:result"/>
    </split>
  </route>
</camelContext>

<bean id="threadPoolExecutor" class="java.util.concurrent.ThreadPoolExecutor">
  <constructor-arg index="0" value="8"/>
  <constructor-arg index="1" value="16"/>
  <constructor-arg index="2" value="0"/>
  <constructor-arg index="3" value="MILLISECONDS"/>
  <constructor-arg index="4"><bean class="java.util.concurrent.LinkedBlockingQueue"/></constructor-arg>
</bean>

由于分割器 可以使用任何 表达式进行分割，因此您可以通过调用 method （） 表达式来使用 bean 执行分割。bean 应返回可迭代的值，如 java.util.Collection、java.util.Iterator 或数组。

以下路由定义了一个 method （） 表达式，它调用 mySplitterBean bean 实例的方法：

from("direct:body")
        // here we use a POJO bean mySplitterBean to do the split of the payload
        .split()
        .method("mySplitterBean", "splitBody")
        .to("mock:result");
from("direct:message")
        // here we use a POJO bean mySplitterBean to do the split of the message
        // with a certain header value
        .split()
        .method("mySplitterBean", "splitMessage")
        .to("mock:result");

其中 mySplitterBean 是 MySplitterBean 类的实例，它定义如下：

public class MySplitterBean {

    /**
     * The split body method returns something that is iteratable such as a java.util.List.
     *
     * @param body the payload of the incoming message
     * @return a list containing each part split
     */
    public List<String> splitBody(String body) {
        // since this is based on an unit test you can of couse
        // use different logic for splitting as {router} have out
        // of the box support for splitting a String based on comma
        // but this is for show and tell, since this is java code
        // you have the full power how you like to split your messages
        List<String> answer = new ArrayList<String>();
        String[] parts = body.split(",");
        for (String part : parts) {
            answer.add(part);
        }
        return answer;
    }

    /**
     * The split message method returns something that is iteratable such as a java.util.List.
     *
     * @param header the header of the incoming message with the name user
     * @param body the payload of the incoming message
     * @return a list containing each part split
     */
    public List<Message> splitMessage(@Header(value = "user") String header, @Body String body) {
        // we can leverage the Parameter Binding Annotations
        // http://camel.apache.org/parameter-binding-annotations.html
        // to access the message header and body at same time,
        // then create the message that we want, splitter will
        // take care rest of them.
        // *NOTE* this feature requires {router} version >= 1.6.1
        List<Message> answer = new ArrayList<Message>();
        String[] parts = header.split(",");
        for (String part : parts) {
            DefaultMessage message = new DefaultMessage();
            message.setHeader("user", part);
            message.setBody(body);
            answer.add(message);
        }
        return answer;
    }
}

您可以使用带有 Splitter EIP 的BeanIOSplitter 对象来分割大型有效负载，通过使用流模式以避免将整个内容读取在内存中。以下示例演示了如何使用映射文件（从 classpath 加载）来设置 BeanIOSplitter 对象：

BeanIOSplitter splitter = new BeanIOSplitter();
   splitter.setMapping("org/apache/camel/dataformat/beanio/mappings.xml");
   splitter.setStreamName("employeeFile");

    // Following is a route that uses the beanio data format to format CSV data
    // in Java objects:
    from("direct:unmarshal")
        // Here the message body is split to obtain a message for each row:
         .split(splitter).streaming()
         .to("log:line")
         .to("mock:beanio-unmarshal");

以下示例添加了一个错误处理器：

BeanIOSplitter splitter = new BeanIOSplitter();
   splitter.setMapping("org/apache/camel/dataformat/beanio/mappings.xml");
   splitter.setStreamName("employeeFile");
   splitter.setBeanReaderErrorHandlerType(MyErrorHandler.class);
   from("direct:unmarshal")
      .split(splitter).streaming()
      .to("log:line")
      .to("mock:beanio-unmarshal");

在每个分割交换中设置以下属性：

在处理各个部分后，消息部分会被聚合到单个交换中是一种常见模式。要支持此模式，split （） DSL 命令允许您提供一个 AggregationStrategy 对象作为第二个参数。

以下示例演示了如何在处理所有消息片段后使用自定义聚合策略来重新组合分割消息：

from("direct:start")
    .split(body().tokenize("@"), new MyOrderStrategy())
        // each split message is then send to this bean where we can process it
        .to("bean:MyOrderService?method=handleOrder")
        // this is important to end the splitter route as we do not want to do more routing
        // on each split message
    .end()
    // after we have split and handled each message we want to send a single combined
    // response back to the original caller, so we let this bean build it for us
    // this bean will receive the result of the aggregate strategy: MyOrderStrategy
    .to("bean:MyOrderService?method=buildCombinedResponse")

上述路由中使用的自定义聚合策略 MyOrderStrategy 的实现如下：

/**
 * This is our own order aggregation strategy where we can control
 * how each split message should be combined. As we do not want to
 * lose any message, we copy from the new to the old to preserve the
 * order lines as long we process them
 */
public static class MyOrderStrategy implements AggregationStrategy {

    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        // put order together in old exchange by adding the order from new exchange

        if (oldExchange == null) {
            // the first time we aggregate we only have the new exchange,
            // so we just return it
            return newExchange;
        }

        String orders = oldExchange.getIn().getBody(String.class);
        String newLine = newExchange.getIn().getBody(String.class);

        LOG.debug("Aggregate old orders: " + orders);
        LOG.debug("Aggregate new order: " + newLine);

        // put orders together separating by semi colon
        orders = orders + ";" + newLine;
        // put combined order back on old to preserve it
        oldExchange.getIn().setBody(orders);

        // return old as this is the one that has all the orders gathered until now
        return oldExchange;
    }
}

启用并行处理后，理论上可以让后续消息准备好在较早之前进行聚合。换句话说，消息片段可能会达到聚合器没有顺序。默认情况下，这不会发生，因为分割器实施在将消息段重新安排到聚合器之前将其重新安排为原始顺序。

如果您希望在消息就绪后马上聚合消息部分（并可能没有顺序），您可以启用 streaming 选项，如下所示：

from("direct:streaming")
  .split(body().tokenize(","), new MyOrderStrategy())
    .parallelProcessing()
    .streaming()
    .to("activemq:my.parts")
  .end()
  .to("activemq:all.parts");

您还可以提供用于流的自定义迭代器，如下所示：

// Java
import static org.apache.camel.builder.ExpressionBuilder.beanExpression;
...
from("direct:streaming")
     .split(beanExpression(new MyCustomIteratorFactory(),  "iterator"))
     .streaming().to("activemq:my.parts")

如果传入的消息是一个非常大的 XML 文件，您可以在流传输模式中使用令牌ize XML 子命令最高效地处理消息。

例如，如果一个包含 顺序 元素序列的大型 XML 文件，您可以使用类似如下的路由将文件分成 顺序 元素：

from("file:inbox")
  .split().tokenizeXML("order").streaming()
  .to("activemq:queue:order");

您可以通过定义类似如下的路由在 XML 中执行同样的操作：

<route>
  <from uri="file:inbox"/>
  <split streaming="true">
    <tokenize token="order" xml="true"/>
    <to uri="activemq:queue:order"/>
  </split>
</route>

通常，您需要访问在令牌元素的一个界(ancestor)元素中定义的命名空间。您可以将命名空间定义从 ancestor 元素之一复制到 token 元素中，方法是将您要从哪个元素继承命名空间定义。

在 Java DSL 中，您将 ancestor 元素指定为 tokenizeXML 的第二个参数。例如，从 enclosing orders 元素继承命名空间定义：

from("file:inbox")
  .split().tokenizeXML("order", "orders").streaming()
  .to("activemq:queue:order");

在 XML DSL 中，您可以使用 inheritNamespaceTagName 属性指定 ancestor 元素。例如：

<route>
  <from uri="file:inbox"/>
  <split streaming="true">
    <tokenize token="order"
              xml="true"
              inheritNamespaceTagName="orders"/>
    <to uri="activemq:queue:order"/>
  </split>
</route>

split DSL 命令支持以下选项：