camel-rabbitMQ：在单个camel消费者中从多个rabbitmq队列进行消费

Question

我有以下场景：有3个rabbitmq队列，生产者根据消息的优先级将消息推送到这些队列。(myqueue_high，myqueue_medium，myqueue_low）我希望有一个可以按顺序或优先级从这些队列中提取的单一使用者，即只要消息在那里，它就一直从高队列中提取。它是从介质中拉出来的。如果medium也是空的，它从Low拉出。

我如何实现这一点？我需要编写自定义组件吗？

Answer 1

将所有消息放入一个队列，但具有不同的优先级会更容易。这样，优先级排序将在代理中完成，并且Camel使用者将获得已经按优先级排序的消息。但是，RabbitMQ实现了FIFO原则，不支持优先级处理（目前为止）。

解决方案1

Camel允许您使用重新排序器根据某个比较器重新组织消息：https://Camel.apache.org/resequencer.html：

from("rabbitmq://hostname[:port]/myqueue_high")
    .setHeader("priority", constant(9))
    .to("direct:messageProcessing");

from("rabbitmq://hostname[:port]/myqueue_medium")
    .setHeader("priority", constant(5))
    .to("direct:messageProcessing");

from("rabbitmq://hostname[:port]/myqueue_low")
    .setHeader("priority", constant(1))
    .to("direct:messageProcessing");

// sort by priority by allowing duplicates (message can have same priority)
// and use reverse ordering so 9 is first output (most important), and 0 is last
// (of course we could have set the priority the other way around, but this way
// we keep align with the JMS specification...)
// use batch mode and fire every 3th second
from("direct:messageProcessing")
    .resequence(header("priority")).batch().timeout(3000).allowDuplicates().reverse()
    .to("mock:result");

这样，所有传入消息都被路由到相同的子路由(direction:messageprocessing)，在该子路由中，消息根据传入路由设置的priority头重新排序。

解决方案2

将SEDA与优先级队列一起使用：

final PriorityBlockingQueueFactory<Exchange> priorityQueueFactory = new PriorityBlockingQueueFactory<Exchange>();
priorityQueueFactory.setComparator(new Comparator<Exchange>() {
    @Override
    public int compare(final Exchange exchange1, final Exchange exchange2) {
        final Integer prio1 = (Integer) exchange1.getIn().getHeader("priority");
        final Integer prio2 = (Integer) exchange2.getIn().getHeader("priority");
        return -prio1.compareTo(prio2);  // 9 has higher priority then 0
    }
});

final SimpleRegistry registry = new SimpleRegistry();
registry.put("priorityQueueFactory", priorityQueueFactory);

final ModelCamelContext context = new DefaultCamelContext(registry);
// configure and start your context here...

from("rabbitmq://hostname[:port]/myqueue_high")
    .setHeader("priority", constant(9))
    .to("seda:priority?queueFactory=#priorityQueueFactory");  // reference queue in registry

from("rabbitmq://hostname[:port]/myqueue_medium")
    .setHeader("priority", constant(5))
    .to("seda:priority?queueFactory=#priorityQueueFactory");

from("rabbitmq://hostname[:port]/myqueue_low")
    .setHeader("priority", constant(1))
    .to("seda:priority?queueFactory=#priorityQueueFactory");

from("seda:priority")
    .to("direct:messageProcessing");

解决方案3

如果在发生故障时需要持久性，请使用Camel的ActiveMQ组件等JMS而不是SEDA。只需通过设置jmspriperation头，将来自RabbitMQ的传入消息转发到JMS目的地。

解决方案4

完全跳过RabbitMQ，只使用支持优先级的JMS代理，如ActiveMQ。