title: Rabbit Mq java client parallel consumption
body: <p>I want to process messages from a rabbitMq queue in parallel. The queue is configured to be autoAck =false. I am using the camel-rabbitMQ support for <a href="http://camel.apache.org/rabbitmq.html" rel="noreferrer">camel endpoints</a>, which has support for a threadPoolSize parameter, but this does not have the desired effect. Messages are still processed serially off the queue, even when threadpoolsize=20.</p>
  
  <p>From debugging through the code I can see that the threadpoolsize parameter is used to create an ExecutorService that is used to pass to the rabbit connectionfactory as described <a href="http://www.rabbitmq.com/api-guide.html#advanced-connection" rel="noreferrer">here</a>. This all looks good until you get into the rabbit <code>ConsumerWorkService</code>. Here messages are processed in block of max size 16 messages. Each message in a block is processed serially and then if there is more work to do the executor service is invokes with the next block. A code snippet of this is below. From this use of the executor service I can't see how the messages can be processed in parallel. The executorservice only ever has one piece of work to perform at a time.</p>
  
  <p>What am I Missing?</p>
  
  <pre><code>private final class WorkPoolRunnable implements Runnable {
  
          public void run() {
              int size = MAX_RUNNABLE_BLOCK_SIZE;
              List&lt;Runnable&gt; block = new ArrayList&lt;Runnable&gt;(size);
              try {
                  Channel key = ConsumerWorkService.this.workPool.nextWorkBlock(block, size);
                  if (key == null) return; // nothing ready to run
                  try {
                      for (Runnable runnable : block) {
                          runnable.run();
                      }
                  } finally {
                      if (ConsumerWorkService.this.workPool.finishWorkBlock(key)) {
                          ConsumerWorkService.this.executor.execute(new WorkPoolRunnable());
                      }
                  }
              } catch (RuntimeException e) {
                  Thread.currentThread().interrupt();
              }
          }
  </code></pre>