I've tested JMS messages sending serially and concurrently(5 threads send jms messages concurrently from the producer).
When I send 100 messages concurrently, few messages payload at the receiving end are NULL. When sent serially there is no issue.
Do I need to setup a session pool or use MDB at the consumer side to handle the messages concurrently? The setup of the JMS is good, because we are receiving messages. Am I missing anything here?
Short description of the pproject setup:
- Publisher is a stateless session bean
- Weblogic 8.1 jms server connection factory and destination are retrieved through JNDI
- Consumer is a java class which subscribes to this server JMS queue and performs the tasks. (this is not a MDB or a Threaded class, listens to the queue asynchronously)
EDITED
JmsConsumer
package net;
import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Date;
import java.util.Hashtable;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
public class ReadJMS implements MessageListener, ExceptionListener {
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
public final static String PROVIDER_URL = "t3://address:7003";
public final static String JMS_FACTORY = "MSS.QueueConnectionFactory";
public final static String QUEUE = "jms.queue";
@SuppressWarnings("null")
public void receiveMessage() throws Exception {
// System.out.println("receiveMessage()..");
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, PROVIDER_URL);
// Define queue
QueueReceiver qreceiver = null;
QueueSession qsession = null;
QueueConnection qcon = null;
ReadJMS async = new ReadJMS();
try {
InitialContext ctx = new InitialContext(env);
QueueConnectionFactory qconFactory = (QueueConnectionFactory) ctx
.lookup(JMS_FACTORY);
qcon = qconFactory.createQueueConnection();
qcon.setExceptionListener(async);
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = (Queue) ctx.lookup(QUEUE);
qreceiver = qsession.createReceiver(queue);
qreceiver.setMessageListener(async);
qcon.start();
System.out.println("readingMessage()..");
// TextMessage msg = (TextMessage) qreceiver.receive();
// System.out.println("Message read from " + QUEUE + " : "
// + msg.getText());
// msg.acknowledge();
} catch (Exception ex) {
ex.printStackTrace();
}
// } finally {
// if (qreceiver != null)
// qreceiver.close();
// if (qsession != null)
// qsession.close();
// if (qcon != null)
// qcon.close();
// }
}
public static void main(String[] args) throws Exception {
ReadJMS test = new ReadJMS();
System.out.println("init");
test.receiveMessage();
while (true) {
Thread.sleep(10000);
}
}
public void onException(JMSException arg0) {
System.err.println("Exception: " + arg0.getLocalizedMessage());
}
public synchronized void onMessage(Message arg0) {
try {
if(((TextMessage)arg0).getText() == null || ((TextMessage)arg0).getText().trim().length()==0){
System.out.println(" " + QUEUE + " : "
+ ((TextMessage) arg0).getText());
}
System.out.print(".");
PrintWriter out = new PrintWriter(new BufferedWriter(
new FileWriter("Output.txt", true)));
Date now = new Date();
out.println("message: "+now.toString()+ " - "+((TextMessage)arg0).getText()+"");
out.close();
} catch (JMSException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
}
WriteJms
package net;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Hashtable;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
public class WriteJMS {
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
public final static String PROVIDER_URL = "t3://url:7003";
public final static String JMS_FACTORY = "MSS.QueueConnectionFactory";
public final static String QUEUE = "jms.queue";
@SuppressWarnings("unchecked")
public void sendMessage() throws Exception {
@SuppressWarnings("rawtypes")
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, PROVIDER_URL);
// Define queue
QueueSender qsender = null;
QueueSession qsession = null;
QueueConnection qcon = null;
try {
InitialContext ctx = new InitialContext(env);
QueueConnectionFactory qconFactory = (QueueConnectionFactory) ctx
.lookup(JMS_FACTORY);
qcon = qconFactory.createQueueConnection();
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = (Queue) ctx.lookup(QUEUE);
TextMessage msg = qsession.createTextMessage();
msg.setText("<eventMessage><eventId>123</eventId><eventName>123</eventName><documentNumber>123</documentNumber><customerId>123</customerId><actDDTaskDate>123</actDDTaskDate><taskStatusErrorMessage>123</taskStatusErrorMessage></eventMessage>");
qsender = qsession.createSender(queue);
qsender.send(msg);
System.out.println("Message [" + msg.getText()
+ "] sent to Queue: " + QUEUE);
} catch (Exception ex) {
ex.printStackTrace();
} finally {
if (qsender != null)
qsender.close();
if (qsession != null)
qsession.close();
if (qcon != null)
qcon.close();
}
}
}