8.4. Splitter

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

图 8.4. Splitter Pattern

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

在拆分器将原始消息分成部分之前，它会制作原始消息的应复制。在 shouldow copy 中，原始消息的标头和有效负载仅作为参考复制。虽然拆分器本身不会将生成的消息部分路由到不同的端点，但拆分消息的部分可能会推向二级路由。

由于消息部分应该浏览副本，所以它们仍然与原始消息相关联。因此，不能独立修改它们。如果要将自定义逻辑应用到消息部分的不同副本，然后将其路由到一组端点，则必须使用 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");
    }
};

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

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 表达式，使用令牌来拆分正文或标头，其中使用 tokenize 元素定义 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 （） 子句将拆分器不一次读取整个文件（如果文件较大，则性能会大大提高）。

相同的路由可以在 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>

如果进入拆分人的交换具有 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>

由于拆分器 可以使用任何 表达式来拆分，您可以使用 bean 执行拆分（通过调用 method （） 表达式）。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 对象来分割大型有效负载，以避免将所有内容读取到内存中。以下示例演示了如何使用映射文件来设置 BeanIOSplitter 对象，该文件是从 classpath 加载的：

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

启用并行处理后，理论上可以将后续消息片段在之前的聚合之前准备好进行聚合。换句话说，消息部分可能会按顺序到达聚合器。默认情况下，这不会发生，因为分割器的实现会将消息片段重新排列回其原始顺序，然后再将它们传递至聚合器。

如果您希望在消息片段就绪后尽快聚合（可能按顺序出现），您可以启用流传输选项，如下所示：

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 文件，您可以在流传输模式中使用 tokenizeXML 子命令，最高效地处理消息。

例如，如果一个包含一系列 顺序 元素的大型 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>

通常，您需要访问令牌元素中的一个所含（级）元素中定义的命名空间。您可以通过指定您要从命名空间定义继承命名空间定义，将命名空间定义从令牌元素复制到 token 元素中。

在 Java DSL 中，您要将 ancestor 元素指定为 tokenizeXML 的第二个参数。例如，从包含的 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 命令支持以下选项：