与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

您应该使用concurrentConsumers选项在评论中说，

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

通知也有maxConcurrentConsumers您可以设置为使用最小/最大范围的并发消费者的，所以骆驼会自动生长/取决于收缩负载。

看到JMS文档的详细信息在

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