You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@activemq.apache.org by Naresh Bhatia <NB...@sapient.com> on 2007/06/20 02:44:23 UTC
How to receive messages concurrently from a queue?
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
--
View this message in context: http://www.nabble.com/How-to-receive-messages-concurrently-from-a-queue--tf3949698s2354.html#a11205460
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: How to receive messages concurrently from a queue?
Posted by Naresh Bhatia <NB...@sapient.com>.
Great! Thanks.
Naresh
--
View this message in context: http://www.nabble.com/How-to-receive-messages-concurrently-from-a-queue--tf3949698s2354.html#a11215154
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: How to receive messages concurrently from a queue?
Posted by Jonas Lim <jl...@exist.com>.
Hi Naresh,
I added a bit of documentation on the default values here
http://cwiki.apache.org/confluence/display/ACTIVEMQ/What+is+the+prefetch+limit+for
Regards,
Jonas
Naresh Bhatia wrote:
> Jonas, that did the trick. Thank you.
>
> BTW, are the default values documented somewhere? I could only find them
> burried deep in the JMX console.
>
> Naresh
>
Re: How to receive messages concurrently from a queue?
Posted by Naresh Bhatia <NB...@sapient.com>.
Jonas, that did the trick. Thank you.
BTW, are the default values documented somewhere? I could only find them
burried deep in the JMX console.
Naresh
--
View this message in context: http://www.nabble.com/How-to-receive-messages-concurrently-from-a-queue--tf3949698s2354.html#a11213111
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: How to receive messages concurrently from a queue?
Posted by Jonas Lim <jl...@exist.com>.
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
>