activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jonas Lim <j...@exist.com>
Subject Re: How to receive messages concurrently from a queue?
Date Wed, 20 Jun 2007 07:01:41 GMT
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
>   

Mime
View raw message