11.4. 竞争消费者

图 11.3 “竞争消费者模式” 中显示的 竞争消费者 模式使多个消费者能够从同一队列拉取信息，并保证 每个消息仅消耗一次。此模式可用于将串行消息处理替换为并发消息处理（响应延迟出现相应的减少）。

图 11.3. 竞争消费者模式

以下组件演示了竞争消费者模式：

定期的 JMS 队列会隐式保证每个消息一次只能被使用。因此，JMS 队列自动支持竞争消费者模式。例如，您可以定义三个竞争消费者，从 JMS 队列 HighVolumeQ 中拉取消息，如下所示：

from("jms:HighVolumeQ").to("cxf:bean:replica01");
from("jms:HighVolumeQ").to("cxf:bean:replica02");
from("jms:HighVolumeQ").to("cxf:bean:replica03");

其中 CXF (Web 服务)端点、replica 01、replica02 和 replica03，请并行处理来自 HighVolumeQ 队列的消息。

或者，您可以设置 JMS 查询选项 concurrentConsumers，以创建竞争消费者的线程池。例如，以下路由会创建一个三个竞争线程池，该池从指定队列中选择消息：

from("jms:HighVolumeQ?concurrentConsumers=3").to("cxf:bean:replica01");

并且 concurrentConsumers 选项也可以在 XML DSL 中指定，如下所示：

 <route>
   <from uri="jms:HighVolumeQ?concurrentConsumers=3"/>
   <to uri="cxf:bean:replica01"/>
 </route>

SEDA 组件的目的是通过将计算划分为阶段来简化并发处理。SEDA 端点基本上封装内存阻塞队列（由 java.util.concurrent.BlockingQueue实施）。因此，您可以使用 SEDA 端点将路由分成阶段，其中每个阶段可能使用多个线程。例如，您可以定义一个由两个阶段组成的 SEDA 路由，如下所示：

// Stage 1: Read messages from file system.
from("file://var/messages").to("seda:fanout");

// Stage 2: Perform concurrent processing (3 threads).
from("seda:fanout").to("cxf:bean:replica01");
from("seda:fanout").to("cxf:bean:replica02");
from("seda:fanout").to("cxf:bean:replica03");

第一个阶段包含一个线程，它会消耗来自文件端点 file://var/messages 并将其路由到 SEDA 端点 seda:fanout。第二个阶段包含三个线程：一个线程将交换到 cxf:bean:replica01、路由交换到 cxf:bean:replica02 的线程，以及路由交换到 cxf:bean:replica03 的线程。这三个线程竞争为从 SEDA 端点获取交换实例，该端点使用阻塞队列来实施。由于阻塞队列使用锁定来防止多个线程每次访问队列，因此您可以保证每个交换实例只能消耗一次。

有关 SEDA 端点和 thread （） 创建的线程池之间的区别，请参阅 NameOfCamelCompRef 中的 SEDA 组件。