1.5. 处理器

概述

要让路由器做比将消费者端点连接到制作者端点相比，您可以向路由添加 处理器 更为有趣。处理器是一个命令，您可以插入到路由规则中，以执行通过该规则执行流的任意处理。Apache Camel 提供各种不同处理器，如表 1.1 “Apache Camel Processors” 所示。

表 1.1. Apache Camel Processors
Java DSL	XML DSL	描述
`aggregate()`	`aggregate`	第 8.5 节 “聚合器”: 创建一个聚合器，它将多个传入交换组合成一个交换。
`aop()`	`AOP`	使用 Aspect Oriented Programming (AOP)在指定子路由之前和之后工作。
`bean （）`, `beanRef （）`	`bean`	通过调用 Java 对象（或 bean）的方法来处理当前的交换。请参阅第 2.4 节 “Bean 集成”。
`choice()`	`choice`	第 8.1 节 “基于内容的路由器”：使用 when 和 `otherwise` 子句，根据交换内容 `选择特定的` 子路由。
`convertBodyTo()`	`convertBodyTo`	将 In 消息正文转换为指定的类型。
`delay()`	`delay`	第 8.9 节 “Delayer”: 将交换的传播延迟到路由的后部分。
`doTry()`	`doTry`	创建用于处理异常的尝试/分散块，使用 `doCatch`、`do lastly`、和 `end` 子句。
`end()`	N/A	结束当前命令块。
`enrich()`,`enrichRef()`	`enrich`	第 10.1 节 “内容增强”：将当前的交换与指定制作者端点 URI 请求的数据合并。
`filter()`	`filter`	第 8.2 节 “消息过滤器”: 使用 predicate 表达式来过滤传入的交换。
`idempotentConsumer()`	`idempotentConsumer`	第 11.8 节 “幂等的消费者”：实施策略以抑制重复消息。
`inheritErrorHandler()`	`@inheritErrorHandler`	布尔值选项，可用于在特定路由节点上禁用继承的错误处理程序（定义为 Java DSL 中的子使用和 XML DSL 中的属性）。
`inOnly()`	`inOnly`	将当前交换的 MEP 设置为仅限（如果没有参数），或者将交换作为仅限给指定的端点。
`inOut()`	`inOut`	将当前交换的 MEP 设置为 InOut （如果没有参数），或者将交换作为 InOut 发送到指定的端点。
`loadBalance()`	`loadBalance`	第 8.10 节 “Load Balancer”：对端点集合实施负载平衡。
`log()`	`log`	将消息记录到控制台。
`loop()`	`loop`	第 8.16 节 “循环”：重复将每个交换重新放置到路由的后部分。
`markRollbackOnly()`	`@markRollbackOnly`	（事务）标记仅回滚的当前事务（不会引发异常）。在 XML DSL 中，此选项被设置为 `回滚` 元素上的布尔值属性。请参阅 Apache Karaf 事务指南。
`markRollbackOnlyLast()`	`@markRollbackOnlyLast`	（事务）如果之前已与该线程关联了一个或多个事务，然后暂停，这个命令会标记仅回滚的最新事务（不会引发异常）。在 XML DSL 中，此选项被设置为 `回滚` 元素上的布尔值属性。请参阅 Apache Karaf 事务指南。
`marshal()`	`marshal`	使用指定的数据格式转换为低级或二进制格式，以准备通过特定的传输协议发送。
`multicast()`	`multicast`	第 8.13 节 “多播”：将当前交换到多个目的地，每个目的地获得自己的交换副本。
`onCompletion()`	`onCompletion`	定义一个子路由（以 Java DSL `结尾` ）在主路由完成后执行该子路由。另请参阅第 2.14 节 “OnCompletion”。
`onException()`	`onException`	定义 Java DSL 中的 `end （）` 终止的子路由（通过 Java DSL 结尾），每当指定的异常发生时执行。通常在其自己的行中定义（不在路由中）。
`pipeline()`	`pipeline`	第 5.4 节 “管道和过滤器”: 将交换发送到一系列端点，其中一个端点的输出将成为下一个端点的输入。另请参阅第 2.1 节 “Pipeline 处理”。
`policy()`	`policy`	将策略应用到当前路由（目前仅用于事务策略事项处理器）中，请参阅 Apache Karaf 事务指南。
`pollEnrich()`,`pollEnrichRef()`	`pollEnrich`	第 10.1 节 “内容增强”: 将当前交换与来自指定的消费者端点 URI 的数据进行合并。
`process()`,`processRef`	`process`	在当前交换上执行自定义处理器。请参阅 “自定义处理器”一节和第 III 部分 “高级 Camel 编程”。
`recipientList()`	`recipientList`	第 8.3 节 “接收者列表”: 将交换发送到运行时计算的接收者列表（例如，基于标头的内容）。
`removeHeader()`	`removeHeader`	从交换的 In 消息中删除指定的标头。
`removeHeaders()`	`removeHeaders`	从交换的 In 消息中删除与指定模式匹配的标头。这个模式可以采用 `前缀\*`
`removeProperty()`	`removeProperty`	从交换删除指定的交换属性。
`removeProperties()`	`removeProperties`	从交换中删除与指定模式匹配的属性。以逗号分隔的 1 或多个字符串列表作为参数。第一个字符串是模式（请参阅上面的 `删除Headers （）` ）。后续字符串指定例外 - 这些属性保留。
`resequence()`	`resequence`	第 8.6 节 “Resequencer”: 根据指定比较器操作对传入交换进行重新排序。支持批处理模式和流模式。
`rollback()`	`rollback`	（事务）标记仅进行回滚的当前事务（此外，默认情况下也会增加异常）。请参阅 Apache Karaf 事务指南。
`routingSlip()`	`routingSlip`	第 8.7 节 “路由 Slip”：根据从 slip 标头中提取的端点 URI 列表，通过构建的管道路由交换。
`sample()`	`sample`	创建一个抽样限制，允许您从路由上的流量中提取交换示例。
`setBody()`	`setBody`	设置交换的 In 消息的消息正文。
`setExchangePattern()`	`setExchangePattern`	将当前交换的 MEP 设置为指定的值。请参阅 “消息交换模式”一节。
`setHeader()`	`setHeader`	在交换的 In 消息中设置指定的标头。
`setOutHeader()`	`setOutHeader`	设置交换的 Out 消息中指定的标头。
`setProperty()`	`setProperty()`	设置指定的交换属性。
`sort()`	`排序`	对 In message body 的内容进行排序（其中可以选择性地指定自定义比较器）。
`split()`	`split`	第 8.4 节 “Splitter”：将当前交换分成一系列交换，每个交换都包含原始消息正文的片段。
`stop()`	`stop`	停止路由当前交换并将其标记为已完成。
`threads()`	`threads`	创建一个线程池，用于对路由的后一部分进行并发处理。
`throttle()`	`throttle`	第 8.8 节 “Throttler”: 将流率限制为指定级别（每秒交换数）。
`throwException()`	`throwException`	抛出指定的 Java 异常.
`to()`	`至`	将交换发送到一个或多个端点。请参阅第 2.1 节 “Pipeline 处理”。
`toF()`	N/A	使用字符串格式将交换发送到端点。也就是说，端点 URI 字符串可以嵌入 C `printf （）` 函数方式中的替换。
`transacted()`	`transacted`	创建一个 Spring 事务范围，其中包含该路由的后一部分。请参阅 Apache Karaf 事务指南。
`transform()`	`翻译`	第 5.6 节 “消息转换器”: 将 In message 标头复制到 Out message 标头，并将 Out 消息正文设置为指定的值。
`unmarshal()`	`unmarshal`	使用指定的数据格式将 In 消息正文从低级或二进制格式转换为高级别的格式。
`validate()`	`validate`	取一个 predicate 表达式来测试当前消息是否有效。如果 predicate 返回 `false`，则引发 `PredicateValidationException` 异常。
`wireTap()`	`wireTap`	第 12.3 节 “wire Tap”: 使用 `ExchangePattern.InOnly` MEP，将当前交换的副本发送到指定的有线 tap URI。

有些示例处理器

要了解如何在路由中使用处理器，请查看以下示例：

选择

choice （） 处理器是一个条件语句，用于将传入消息路由到备选制作者端点。每个替代制作者端点的前面是一个 when （） 方法，该方法采用 predicate 参数。如果 predicate 为 true，则会选择以下目标，否则处理规则中的下一个 when （） 方法。例如，以下 choice （） 处理器将传入的信息定向到 Target1、Target 2 或 Target3，具体取决于 Predicate1 和 Predicate2 的值：

from("SourceURL")
    .choice()
        .when(Predicate1).to("Target1")
        .when(Predicate2).to("Target2")
        .otherwise().to("Target3");

在 Spring XML 中，或等效于 Spring XML 中：

<camelContext id="buildSimpleRouteWithChoice" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="SourceURL"/>
    <choice>
      <when>
        <!-- First predicate -->
        <simple>header.foo = 'bar'</simple>
        <to uri="Target1"/>
      </when>
      <when>
        <!-- Second predicate -->
        <simple>header.foo = 'manchu'</simple>
        <to uri="Target2"/>
      </when>
      <otherwise>
        <to uri="Target3"/>
      </otherwise>
    </choice>
  </route>
</camelContext>

在 Java DSL 中，有一个特殊的情形，您可能需要使用 endChoice （） 命令。某些标准 Apache Camel 处理器允许您使用特殊子使用指定额外参数，从而有效地打开额外的嵌套级别，该嵌套通常由 end （） 命令终止。例如，您可以将负载均衡器子指定为 loadBalance （）.roundRobin （）.to ("mock:foo").to ("mock:bar").end （），用于平衡 mock:foo 和 mock:bar 间的消息。但是，如果负载均衡器子嵌入在选择条件中，则需要使用 endChoice （） 命令终止子句，如下所示：

from("direct:start")
    .choice()
        .when(bodyAs(String.class).contains("Camel"))
            .loadBalance().roundRobin().to("mock:foo").to("mock:bar").endChoice()
        .otherwise()
            .to("mock:result");

Filter

可以使用 filter （） 处理器来防止不间断消息到达制作者端点。它采用单个 predicate 参数：如果 predicate 为 true，则允许消息交换到制作者；如果 predicate 为 false，则消息交换会被阻断。例如，以下过滤器会阻断一个消息交换，除非传入的消息包含一个标头 foo，其值等于 bar ：

from("SourceURL").filter(header("foo").isEqualTo("bar")).to("TargetURL");

在 Spring XML 中，或等效于 Spring XML 中：

<camelContext id="filterRoute" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="SourceURL"/>
    <filter>
      <simple>header.foo = 'bar'</simple>
      <to uri="TargetURL"/>
    </filter>
  </route>
</camelContext>

Throttler

throttle （） 处理器确保制作者端点不会过载。throttler 的工作原理是限制每秒可传递的消息数量。如果传入的消息超过指定率，则节流者会累积缓冲区中过量消息，并将它们传输速度会更慢于制作者端点。例如，要将吞吐量率限制为每秒 100 个信息，您可以定义以下规则：

from("SourceURL").throttle(100).to("TargetURL");

在 Spring XML 中，或等效于 Spring XML 中：

<camelContext id="throttleRoute" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="SourceURL"/>
    <throttle maximumRequestsPerPeriod="100" timePeriodMillis="1000">
      <to uri="TargetURL"/>
    </throttle>
  </route>
</camelContext>

自定义处理器

如果此处描述的标准处理器不提供您需要的功能，您总是可以自行定义自定义处理器。要创建自定义处理器，可定义一个实施 org.apache.camel.Processor 接口的类，并覆盖 process （） 方法。以下自定义处理器 MyProcessor 会从传入信息中删除名为 foo 的标头：

例 1.3. 实施自定义处理器类

public class MyProcessor implements org.apache.camel.Processor {
public void process(org.apache.camel.Exchange exchange) {
  inMessage = exchange.getIn();
  if (inMessage != null) {
      inMessage.removeHeader("foo");
  }
}
};

要将自定义处理器插入到路由器规则中，调用 process （） 方法，该方法为将处理器插入到规则的通用机制。例如，以下规则调用在例 1.3 “实施自定义处理器类” 中定义的处理器：

org.apache.camel.Processor myProc = new MyProcessor();

from("SourceURL").process(myProc).to("TargetURL");

概述

有些示例处理器

选择

Filter

Throttler

自定义处理器

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Red Hat legal and privacy links

Red Hat legal and privacy links