ActiveMQ Producer Multiple Queues One Session

Question

How is it possible to enqueue messages to different queues using a persistent connection / session in ActiveMQ?

What I have done:

public class ActiveMQProducer {

    private static final Logger LOGGER = Logger.getLogger(ActiveMQProducer.class);
    private Connection connection;
    private MessageProducer producer;
    private Session session;
    String activeMQConnection;

    public ActiveMQProducer() throws ConfigurationException, JMSException {
        activeMQConnection = ActiveMQPropertyManagerFactory.getInstance().getString("active.mq.url");
    }

    public void setupActiveMQ(String queueName) throws JMSException {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(activeMQConnection);
        factory.setRejectedTaskHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        connection = factory.createConnection();
        connection.start();

        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(queueName);
        producer = session.createProducer(queue);

    }

    public void getConnection(String queueName) throws JMSException {

        if (connection == null || session == null) {
            Object object = new Object();
            synchronized (object) {
                setupActiveMQ(queueName);
            }
        }
    }

    public <T extends Serializable> T sendToActiveMQ(String queueName, T t) throws JMSException {
        getConnection(queueName);
        ObjectMessage message = session.createObjectMessage(t);
        producer.send(message);
        return null;
    }

    public void sendMessageToActiveMQ(String queueName, String message) throws JMSException {
        getConnection(queueName);
        TextMessage toSend = session.createTextMessage(message);
        producer.send(toSend);
    }
}

I have realized by using this and sending a message to a different queue eventually ActiveMQ runs out of connections because I never close the connection or session:

org.apache.activemq.transport.tcp.ExceededMaximumConnectionsException: Exceeded the maximum number of allowed client connections.

What would be the proper way to handle this? I have about 5 queues I have to send different messages to, should I open a new connection, enqueue and close the connection, is there anyway to keep the session / connections persistent?

Thanks.

Answer 1

here some solutions :

1 Producer per destination :

import java.io.Serializable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ConfigurationException;
import org.apache.log4j.Logger;

public class ActiveMQProducer {

    private static final Logger LOGGER = Logger.getLogger(ActiveMQProducer.class);
    private Connection connection;
    private Session session;
    String activeMQConnection;
    Map<String, MessageProducer> producers = Collections.synchronizedMap(new HashMap<String, MessageProducer>());
    Thread shutdownHook = new Thread(new Runnable() {
        @Override
        public void run() {
            close();
        }
    });

    public ActiveMQProducer() throws ConfigurationException, JMSException {
        activeMQConnection = ActiveMQPropertyManagerFactory.getInstance().getString("active.mq.url");
        setupActiveMQ();
        Runtime.getRuntime().addShutdownHook(shutdownHook);
    }

    public void setupActiveMQ() throws JMSException {
        close();
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(activeMQConnection);
        factory.setRejectedTaskHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    @Override
    protected void finalize() throws Throwable {
        close();
        super.finalize();
    }

    public void close() {
        if (connection != null) {
            try {
                connection.close();
            } catch (Exception e) {
            }
            connection = null;
        }
    }

    public void getConnection() throws JMSException {
        if (connection == null || session == null) {
            setupActiveMQ();
        }
    }

    public MessageProducer getProducer(String queueName) throws JMSException {
        getConnection();
        MessageProducer producer = producers.get(queueName);
        if (producer == null) {
            Queue queue = session.createQueue(queueName);
            producer = session.createProducer(queue);
            producers.put(queueName, producer);
        }
        return producer;
    }

    public <T extends Serializable> T sendToActiveMQ(String queueName, T t) throws JMSException {
        MessageProducer producer = getProducer(queueName);
        ObjectMessage message = session.createObjectMessage(t);
        producer.send(message);
        return null;
    }

    public void sendMessageToActiveMQ(String queueName, String message) throws JMSException {
        MessageProducer producer = getProducer(queueName);
        TextMessage toSend = session.createTextMessage(message);
        producer.send(toSend);
    }
}

1 Producer for all destinations :

import java.io.Serializable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ConfigurationException;
import org.apache.log4j.Logger;

public class ActiveMQProducer2 {

    private static final Logger LOGGER = Logger.getLogger(ActiveMQProducer2.class);
    private Connection connection;
    private Session session;
    String activeMQConnection;
    Map<String, Destination> destinations = Collections.synchronizedMap(new HashMap<String, Destination>());
    private MessageProducer producer;
    Thread shutdownHook = new Thread(new Runnable() {
        @Override
        public void run() {
            close();
        }
    });

    public ActiveMQProducer2() throws ConfigurationException, JMSException {
        activeMQConnection = ActiveMQPropertyManagerFactory.getInstance().getString("active.mq.url");
        setupActiveMQ();
        Runtime.getRuntime().addShutdownHook(shutdownHook);
    }

    public void setupActiveMQ() throws JMSException {
        close();
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(activeMQConnection);
        factory.setRejectedTaskHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(session.createTemporaryQueue());
    }

    @Override
    protected void finalize() throws Throwable {
        close();
        super.finalize();
    }

    public void close() {
        if (connection != null) {
            try {
                connection.close();
            } catch (Exception e) {
            }
            connection = null;
        }
    }

    public void getConnection() throws JMSException {
        if (connection == null || session == null) {
            setupActiveMQ();
        }
    }

    public Destination getDestination(String queueName) throws JMSException {
        getConnection();
        Destination destination = destinations.get(queueName);
        if (destination == null) {
            destination = session.createQueue(queueName);
            destinations.put(queueName, destination);
        }
        return destination;
    }

    public <T extends Serializable> T sendToActiveMQ(String queueName, T t) throws JMSException {
        Destination destination = getDestination(queueName);
        ObjectMessage message = session.createObjectMessage(t);
        producer.send(destination, message);
        return null;
    }

    public void sendMessageToActiveMQ(String queueName, String message) throws JMSException {
        Destination destination = getDestination(queueName);
        TextMessage toSend = session.createTextMessage(message);
        producer.send(destination, toSend);
    }
}