8.6. Resequencer

resequencer 模式（如 图 8.7 “重新排序模式” ）可让您根据顺序表达式重新排序信息。为 sequencing 表达式生成 low 值的消息将移到生成高值的批处理和消息前面的消息。

图 8.7. 重新排序模式

Apache Camel 支持两种重新排序算法：

默认情况下，resequencer 不支持重复消息，在消息到达相同的消息时，将仅保留最后一条消息。但是，在批处理模式下，您可以启用重新排序以允许重复。

默认启用批处理重新排序算法。例如，根据 TimeStamp 标头中包含的时间戳值重新排序传入的消息，您可以在 Java DSL 中定义以下路由：

from("direct:start").resequence(header("TimeStamp")).to("mock:result");

默认情况下，通过收集所有传入消息，以间隔 1000 毫秒（默认 批处理超时），最多获取批处理信息（默认 批处理大小）。您可以通过附加 batch（） DSL 命令来自定义批处理超时和批处理大小的值，该命令使用 BatchResequencerConfig 实例作为唯一参数。例如，若要修改前面的路由，使批处理由 4000 毫秒时间内收集的消息组成，最多 300 个消息，您可以定义 Java DSL 路由，如下所示：

import org.apache.camel.model.config.BatchResequencerConfig;

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
        from("direct:start").resequence(header("TimeStamp")).batch(new BatchResequencerConfig(300,4000L)).to("mock:result");
    }
};

您还可以使用 XML 配置指定批处理重新排序模式。以下示例定义了批处理大小为 300 的批处理重新排序，批处理超时为 4000 毫秒：

<camelContext id="resequencerBatch" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start" />
    <resequence>
      <!--
        batch-config can be omitted for default (batch) resequencer settings
      -->
      <batch-config batchSize="300" batchTimeout="4000" />
      <simple>header.TimeStamp</simple>
      <to uri="mock:result" />
    </resequence>
  </route>
</camelContext>

表 8.2 “批处理重新排序选项” 显示仅在批处理模式下可用的选项。

例如，如果要根据 JMSPriority 重新排序来自 JMS 队列的消息，则需要组合选项、allowDuplicat 和 reverse，如下所示：

from("jms:queue:foo")
        // sort by JMSPriority by allowing duplicates (message can have same JMSPriority)
        // and use reverse ordering so 9 is first output (most important), and 0 is last
        // use batch mode and fire every 3th second
        .resequence(header("JMSPriority")).batch().timeout(3000).allowDuplicates().reverse()
        .to("mock:result");

要启用流重新排序算法，您必须将 stream（） 附加到 resequence（） DSL 命令中。例如，根据 seqnum 标头中的序列号值重新排序传入的信息，您需要定义一个 DSL 路由，如下所示：

from("direct:start").resequence(header("seqnum")).stream().to("mock:result");

流处理重新排序算法基于消息流中空缺的检测，而不是固定批量大小。差错检测与超时相结合，无需事先知道序列数（即批处理大小）的约束。消息必须包含唯一序列号，即 predecessor 和 a successor is known。例如，带有序列号 3 的消息带有序列号 2 的前身消息，以及序列号为 4 的后序消息。由于缺少 3 的后续情况，消息序列 2,3,5 有差距。因此，resequencer 必须保留消息 5，直到消息 4 到达（或超时发生）。

默认情况下，流 resequencer 配置超时为 1000 毫秒，最大消息容量为 100。要自定义流的超时和消息容量，您可以将 StreamResequencerConfig 对象作为参数传递到 stream（）。例如，若要将消息容量为 5000 且超时为 4000 毫秒配置流重新排序器，您需要定义路由，如下所示：

// Java
import org.apache.camel.model.config.StreamResequencerConfig;

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
        from("direct:start").resequence(header("seqnum")).
            stream(new StreamResequencerConfig(5000, 4000L)).
            to("mock:result");
    }
};

如果成功消息之间的最大时间延迟（即，消息流中带有相邻顺序的消息）是已知的，则 resequencer 的 timeout 参数应设置为这个值。在这种情况下，您可以保证流中的所有信息都会被正确传送到下一个处理器。与不不透明时间的差值相比，超时值越低，可能更有可能将消息传送出序列。大型超时值应该获得足够高的容量值的支持，其中使用 capacity 参数以防止重新排序内存不足。

如果要使用 长期 某些类型的序列号，则必须定义自定义比较器，如下所示：

// Java
ExpressionResultComparator<Exchange> comparator = new MyComparator();
StreamResequencerConfig config = new StreamResequencerConfig(5000, 4000L, comparator);
from("direct:start").resequence(header("seqnum")).stream(config).to("mock:result");

您还可以使用 XML 配置指定流重新排序模式。以下示例定义了消息容量为 5000 且超时为 4000 毫秒的流重新排序：

<camelContext id="resequencerStream" xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="direct:start"/>
    <resequence>
      <stream-config capacity="5000" timeout="4000"/>
      <simple>header.seqnum</simple>
      <to uri="mock:result" />
    </resequence>
  </route>
</camelContext>

如果由于某种原因无法评估 sequencing 表达式，requencer EIP 会抛出 CamelExchangeException 异常。如果传入交换无效，如果由于某种原因无法评估不同的表达式。您可以使用 ignoreInvalidExchanges 选项忽略这些异常，这意味着重新排序器会跳过任何无效的交换。

from("direct:start")
  .resequence(header("seqno")).batch().timeout(1000)
    // ignore invalid exchanges (they are discarded)
    .ignoreInvalidExchanges()
  .to("mock:result");

rejectOld 选项可用于防止以任何顺序发送的消息，而不考虑用于重新排序消息的机制。启用 rejectOld 选项时，重新排序器拒绝传入的消息（通过抛出 MessageRejectedException 异常），如果传入的消息是 旧的 （由当前比较器定义）而不是上次发送的消息。

from("direct:start")
    .onException(MessageRejectedException.class).handled(true).to("mock:error").end()
    .resequence(header("seqno")).stream().timeout(1000).rejectOld()
    .to("mock:result");