-1

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:

  1. Publisher is a stateless session bean
  2. Weblogic 8.1 jms server connection factory and destination are retrieved through JNDI
  3. 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();
        }
    }


}
6
  • What do you mean when you say are sent null? Are some messages received have payload null? Commented Jun 27, 2014 at 15:50
  • @AniketThakur Yes, the payload is null.
    – Zeus
    Commented Jun 27, 2014 at 16:04
  • @AniketThakur the message at the receiving end are null. When sending, I'm verifying the logs, it seems to send without any issue. both before and after sending the message
    – Zeus
    Commented Jun 27, 2014 at 16:46
  • @jtahlborn updated with the code
    – Zeus
    Commented Jun 27, 2014 at 17:19
  • um, the sender is the interesting bit.
    – jtahlborn
    Commented Jun 27, 2014 at 17:27

1 Answer 1

-1

Reusing a session across multiple threads is notoriously forbidden. But you don't do this. You create everything (connection session and producer) anew for each message. That is inefficient, but not incorrect and should not cause these errors. The code you give us looks good to me.

I am a little surprised that no exceptions occur at the sending side. Can you give some more details about the JMS implementation? Perhaps there's more information in the message broker's log?

Have you counted the messages and does the number received equal the amount sent? Could someone else be sending messages to the same queue?

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.