使用 ActiveMQ 并行处理多条消息

Question

我想使用简单的处理器/异步处理器作为目标并行处理队列中的消息。处理器每条消息需要一点时间，但可以单独处理每条消息，因此可以同时处理（在正常范围内）。

我很难找到示例，尤其是关于骆驼路线的 xml 配置。

到目前为止，我已经定义了一个线程池、路由和处理器：

<threadPool id="smallPool" threadName="MyProcessorThread" poolSize="5" maxPoolSize="50" maxQueueSize="100"/>
<route>
    <from uri="broker:queue:inbox" />
    <threads executorServiceRef="smallPool">
        <to uri="MyProcessor" />
    </threads>
</route>
<bean id="MyProcessor" class="com.example.java.MyProcessor" />

我的处理器看起来像：

public class MyProcessor implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        Message in = exchange.getIn();
        String msg = in.getBody(String.class);      
        System.out.println(msg);
        try {
            Thread.sleep(10 * 1000); // Do something in the background
        } catch (InterruptedException e) {}
        System.out.println("Done!");
    }
}

不幸的是，当我将消息发布到队列时，它们仍然被一一处理，每个延迟 10 秒（我的“后台任务”）。

谁能指出我使用定义的线程池处理消息的正确方向或解释我做错了什么？

Answer 1

您应该使用 cmets 中所述的 concurrentConsumers 选项，

<route>
    <from uri="broker:queue:inbox?concurrentConsumers=5" />
    <to uri="MyProcessor" />
</route>

请注意，您还可以设置maxConcurrentConsumers 以使用最小/最大范围的并发消费者，因此 Camel 将根据负载自动增长/缩小。

在 JMS 文档中查看更多详细信息

• http://camel.apache.org/jms