How to thread pool a Spring JMS listener

Question

I am setting up a JMS subscriber listener as follows with the goal of achieving a pool of 5 threads listening to topATopic, however, what I see at runtime is multiple consumers processing the same record (recordCount*#of consumers).

I am assuming I am doing something wrong considering I am new to spring.

<bean id="messageListener" class="com.abc.app.mdp.Receiver">
<property name="bean" ref="bean" />
</bean>

<jms:listener-container container-type="default"
connection-factory="connectionFactory" acknowledge="auto" concurrency="5" destination-type="topic" prefetch="1" cache="none" >
<jms:listener destination="topCli_Service" ref="messageListener" 
method="onMessage" subscription="AProjectSubscriber" />
</jms:listener-container>

<bean id="connectionFactory" class="org.springframework.jndi.JndiObjectFactoryB ean">
<property name="jndiName" value="jms/jms-top-notx" />
</bean>

Can somebody please point me in a direction to achieve my goal?

Answer 1

If you want a given message to be consumed by one and only one consumer, you should use a Queue instead of a Topic. A Topic message is broadcast to all available consumers.

Answer 2

Take a look at the concurrency setting on your listener container configuration. The Spring JMS doc suggests that concurrency should be set to 1 for topic listeners. See below.

concurrency: The number of concurrent sessions/consumers to start for each listener. Can either be a simple number indicating the maximum number (e.g. "5") or a range indicating the lower as well as the upper limit (e.g. "3-5"). Note that a specified minimum is just a hint and might be ignored at runtime. Default is 1; keep concurrency limited to 1 in case of a topic listener or if queue ordering is important; consider raising it for general queues.

This post is similar to the rest of your question.

If you need multiple threads to keep up with message volume, your message listener could delegate to a Spring TaskExecutor to process the messages asynchronously. TaskExecutors can be backed by a number of implementations including Thread Pool.