spring boot activemq consumer connection pool

Question

Spring Boot ActiveMQ consumer connection pool is needed to configure? I have only one consumer in spring boot application (as a micro service), producers are in another application. I am little confused by the below: (extracted from http://activemq.apache.org/spring-support.html)

Note: while the PooledConnectionFactory does allow the creation of a collection of active consumers, it does not 'pool' consumers. Pooling makes sense for connections, sessions and producers, which can be seldom-used resources, are expensive to create and can remain idle a minimal cost. Consumers, on the other hand, are usually just created at startup and left going, handling incoming messages as they come. When a consumer is complete, it's preferred to shut down it down rather than leave it idle and return it to a pool for later reuse: this is because, even if the consumer is idle, ActiveMQ will keep delivering messages to the consumer's prefetch buffer, where they'll get held up until the consumer is active again.

At the same page, I can see this:
You can use the activemq-pool org.apache.activemq.pool.PooledConnectionFactory for efficient pooling of the connections and sessions for your collection of consumers, or you can use the Spring JMS org.springframework.jms.connection.CachingConnectionFactory to achieve the same effect

I tried CachingConnectionFactory (which can take ActiveMQConnectionFactory) where it has only few setter to hold cacheConsumers(boolean), cacheProducers(boolean), nothing related to pool the connection. I know that 1 connection can give you multiple session, then per session you have multiple consumer/producer. But my question is for Consumer how do we pool as the above statement is saying leave it to default. So I did this by just one method:
@Bean public JmsListenerContainerFactory myFactory(ConnectionFactory connectionFactory, DefaultJmsListenerContainerFactoryConfigurer configurer) {

    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    // This provides all boot's default to this factory, including the message converter
    factory.setConcurrency("3-10");
    configurer.configure(factory, connectionFactory);
    // You could still override some of Boot's default if necessary.
    return factory;
}</em><br>

Dynamic scaling this link also suggests this, but I could not find concrete solution.

Did someone came across this kind of situation, please give your suggestion. Thanks for reading this post and any help greatly appreciated. Additional details for Production:

This consumer will receive ~500 message per sec.

Using Spring Boot version 1.5.8.RELEASE,

ActiveMQ 5.5 is my JMS

did you find the solution? i'm also confused with the above statements — Vishal Patel

Nir Nir · Accepted Answer · 2017-12-11T09:49:52

There is an package called org.apache.activemq.jms.pool in activemq which provides PooledConsumer. Below is the code for that. Please check and see if it works for you. I know its not the spring way but you can easily manage to customise your poll method.

PooledConnectionFactory pooledConFactory = null;
    PooledConnection pooledConnection = null;
    PooledSession pooledSession = null;
    PooledMessageConsumer pooledConsumer = null;
    Message message = null;
    try
    {
        // Get the connection object from PooledConnectionFactory
        pooledConFactory = ( PooledConnectionFactory ) this.jmsTemplateMap.getConnectionFactory();
        pooledConnection = ( PooledConnection ) pooledConFactory.createConnection();
        pooledConnection.start();

        // Create the PooledSession from pooledConnection object
        pooledSession = ( PooledSession ) pooledConnection.createSession( false, 1 );

        // Create the PooledMessageConsumer from session with given ack mode and destination
        pooledConsumer = ( PooledMessageConsumer ) pooledSession.
                createConsumer( this.jmsTemplateMap.getDefaultDestination(), <messageFilter if any>);

        while ( true )
        {
            message = pooledConsumer.receiveNoWait();
            if ( message != null) 
                break;
        }

    }
    catch ( JMSException ex )
    {
        LOGGER.error("JMS Exception occured, closing the session", ex );
    }
    return message;

spring boot activemq consumer connection pool

1 Answers