JMS Connection delivering messages sent to the queue while the connection was stopped

Question

I am facing an issue with JMS Connection stop() and start(). A simple java program illustrating the same is:

public class Test {
    static Connection producerConn = null;
    static BufferedWriter consumerLog = null;
    static BufferedWriter producerLog = null;

    public static final void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616)");
        producerConn = cf.createConnection();

    producerLog = new BufferedWriter(new FileWriter("produced.log"));
    consumerLog = new BufferedWriter(new FileWriter("consumed.log"));

    new Thread(new Runnable() {
        public void run() {
            try {
                producerConn.start();
                Session session = producerConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Queue queue = session.createQueue("SampleQ1");
                MessageProducer producer = session.createProducer(queue);
                Random random = new Random();
                byte[] messageBytes = new byte[1024];

                for (int i = 0; i < 100; i++) {
                    random.nextBytes(messageBytes);
                    Message message = session.createObjectMessage(messageBytes);
                    producer.send(message);
                    Thread.currentThread().sleep(10);
                    producerLog.write(message.getJMSMessageID());
                    producerLog.newLine();
                    producerLog.flush();
                }
                System.out.println("Produced 100000 messages...");
                producerLog.close();
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }).start();

    System.out.println("Started producer...");


    new Thread(new Runnable() {
        public void run() {
            int count = 0;
            try {
                producerConn.start();
                Session session = producerConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Queue queue = session.createQueue("SampleQ1");
                MessageConsumer consumer = session.createConsumer(queue);
                consumer.setMessageListener(new Test().new MyListener());
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }).start();

    System.out.println("Started consumer...");
}

private class MyListener implements MessageListener{
    private int count = 0;

    public void onMessage(Message message) {
        try {
            message.acknowledge();
            System.out.println("count is " +count++);
            if(count == 5){
                producerConn.stop();
                System.out.println("Sleeping Now for 5 seconds. . ." +System.currentTimeMillis());
                Thread.currentThread().sleep(5000);
                producerConn.start();
            }
            System.out.println("Waking up . . ." +System.currentTimeMillis());
            consumerLog.write(message.getJMSMessageID());
            consumerLog.newLine();
            consumerLog.flush();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}

}

My idea is to simulate the connection stop() and start(). Therefore, in the consumer thread after calling stop(), I have placed a sleep of 5 seconds. However, in the mean time the producer thread continues its job of sending message to the queue.

I expected the test to just consume only the message delivered before the consumer calls stop() and after it calls start() again after waking up from the sleep. But what's happening here is, when consumer wakes up it reads all the messages from the server even those that were sent to the queue when consumer's message reception was stopped.

Am I doing anything wrong here ?

Shashi Shashi · Accepted Answer · 2015-09-09T06:07:56

There is nothing wrong there, it's the correct behavior. In asynchronous messaging producer and consumer are loosely decoupled. A producer does not care whether a consumer is consuming messages or not. It keeps putting messages to a queue while the consumer may be down, or stopped consuming messages or actively consuming messages.

The connection.stop() method has no effect on producer. It affects only consumer, stop() method pauses the delivery of messages from JMS provider to a consumer while start() method starts/resumes message delivery.

Hope this helped.

JMS Connection delivering messages sent to the queue while the connection was stopped

1 Answers