SimpleMessageListenerContainer and TaskExecutor always only 1 message processed

2
votes

I have a SimpleMessageListenerContainer using a TaskExecutor (pool size = 15) that will never process more than 1 message at a time. Here is how it is configured.

SimpleMessageListenerContainer settings are:

concurrentConsumers = 1 (e.g., 1 JMS consumer)
taskExecutor = instance of java.util.concurrent.ThreadPoolExecutor

ThreadPoolExecutor settings are:

corePoolSize = 1
maximumPoolSize= 15
workQueue = LinkedBlockingQueue<Runnable> with size 200

My Expectation:

I expect the 1 JMS consumer to pull off messages and run them as fast as possible in one of the background taskExecutor threads. If 50 messages were on the queue, it would pull off all 50. Run 15 at a time while the other 35 were held in the taskExecutor's LinkedBlockingQueue internal queue.

What's Actually Happening:

Instead, my app processes 1 message at a time. In the JConsole, where I have extended the container and exposed more JMX attributes about the task executor status, I see the "taskExecutor" activeCount is 1 maximum (e.g. ThreadPoolExecutor.getActiveCount()). The ThreadPoolExecutor never moves to 15. So, the JMS Consumer is still processing 1 message at a time even with a taskExecutor with a thread pool of 15 is processing the messages.

Here's what the JConsole JMX readings show after I flooded the queue with tons of messages.

  • ConcurrentConsumers = maps to Spring's SimpleMessageContainer.concurrentConsumers

  • ActiveCount = maps to ThreadPoolExecutor.activeCount

  • MinConcurrentConsumers = maps to ThreadPoolExecutor.corePoolSize
  • MaxConcurrentConsumers = maps to ThreadPoolExecutor.maximumPoolSize
  • InternalQueueCapacity = maps to ThreadPoolExecutor's LinkedBlockingQueue size.

enter image description here

What am I missing? Do I need to use a different implementation of Executor?

1
springspring-jms

1 Answers

0
votes

I found this link helpful in resolving this issue:

how to configure ThreadPoolExecutor to grow/shrink

My corePoolSize was 1 and internally, ThreadPoolExecutor just queued up the messages and processed the queued messages with that single core thread. Setting the corePoolSize higher allowed the growth of the # of threads I wanted.