Re: How to receive messages concurrently from a queue?

[Jonas Lim]

Hi

Can you try setting your prefetch limit to a lower value (ie. 10) ?

http://activemq.apache.org/what-is-the-prefetch-limit-for.html

Regards,
Jonas

Naresh Bhatia wrote:

I have a very simple Java program that creates 10 MessageListeners to receive
JMS messages concurrently from a queue. The MessageListeners share the same
JMS Connection, but create their own Session and MessageConsumer (I have
also tried this with each MessageListener creating its own Connection).
Unfortunately I observe that only the first MessageListener is receiving
messages, none of the others do (even though there are plenty of messages in

the queue). What could I be doing wrong?

Here's the code that starts my MessageListeners:

    for (int i=0; i < numListeners; i++) {
        listeners[i] = new MyMessageListener(jmsTemplate, myQueue);
        listeners[i].start();
    }

Here's the code for my MessageListener:

package samples.jmsqueuesspring.consumer;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsTemplate;

public class MyMessageListener implements MessageListener {

    // ----- Attributes -----
    private Log logger = LogFactory.getLog(MyMessageListener.class);

    /** Number of MyMessageListeners running on this server **/
    private static int listenerCount = 0;

    /** id of this listener */
    private int listenerId;

    // JMS attributes
    private Connection connection;
    private Session session;
    private MessageConsumer consumer;

    // ----- Methods -----
    public MyMessageListener(JmsTemplate jmsTemplate, Destination myQueue)
throws Exception {

        assignId();
        logger.debug("MyMessageListener created: listenerId=" + listenerId);

        // Create a JMS connection, session and consumer
        this.connection =
jmsTemplate.getConnectionFactory().createConnection();
        this.session = connection.createSession(true,
Session.AUTO_ACKNOWLEDGE);
        this.consumer = session.createConsumer(myQueue);
        this.consumer.setMessageListener(this);
    }

    /** Start receiving messages from MyQueue */
    public void start() throws JMSException {
        logger.debug("MyMessageListener started: listenerId=" + listenerId);
        this.connection.start();
    }

    public void close() {
        try {connection.close();} catch (JMSException je) {}
        logger.debug("MyMessageListener closed: listenerId=" + listenerId);
    }

    private synchronized void assignId() {
        this.listenerId = ++listenerCount;
    }

    public void onMessage(Message message) {

        TextMessage textMessage = (TextMessage)message;

        try {
            int messageId = Integer.parseInt(textMessage.getText());

            logger.debug(
                "listner " + listenerId + ": messageId=" + messageId +
                (textMessage.getJMSRedelivered() ? " - Redelivered" : ""));

            this.session.commit();
        }
        catch (Exception e) {
            logger.error("Received Exception", e);
            try {session.rollback();} catch (JMSException je2) {}
        }
    }
}

I am using ActiveMQ 4.1.1.

Thanks.
Naresh