8.6. Resequencer

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

图 8.7. 重新排序器模式

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

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

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

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

默认情况下，通过收集所有传入间隔为 1000 毫秒（默认 批处理超时 ）的所有传入消息来获取批处理，最多 100 个消息（默认 批处理大小）。您可以通过附加 batch （） DSL 命令来自定义批处理超时和批处理大小，该命令使用 BatchResequencerConfig 实例作为其唯一参数。例如，要修改上述路由，以便批处理由 4000 millisecond 时间窗口中收集的消息组成，最多 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 队列重新排序消息，则需要组合选项、allowDuplicates 和 反向，如下所示：

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");

流处理重新排序器算法基于消息流中的差距检测，而不是固定的批处理大小。差距检测与超时相结合，消除了需要提前知道序列消息数（即批处理大小）的约束。消息必须包含一个唯一的序列号，前者和成功者是已知的。例如，带有序列号 3 的消息带有序列号 2，以及序列号为 4 的后续消息。消息序列 23,5 会产生差距，因为缺少 3 的后续者。因此，重新排序器必须保留消息 5，直到消息 4 到达（或超时发生）。

默认情况下，流重新排序器配置为 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");
    }
};

如果消息流中连续消息（即，带有相邻序列号的消息）之间的最大时间延迟已知，则重新排序器的 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>

resequencer EIP 会抛出 CamelExchangeException 异常，如果传入的交换无效是，如果出于某种原因无法评估 sequencing 表达式（例如，因为缺少标头）。您可以使用 ignoreInvalidExchanges 选项忽略这些异常，这意味着重新排序器将跳过任何无效的交换。

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

rejectOld 选项可用于防止消息没有按顺序发送，无论用于重新排序消息的机制是什么。启用 rejectOld 选项后，resequencer 会拒绝传入的消息（通过抛出 MessageRejectedException 异常），如果传入的消息是 旧的 （由当前比较器定义）超过最后发送的消息。

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