JMS（二）：MessageListener，实战Queue ，实战Topic，消息头

MessageListener

消息的消费者接收消息可以采用两种方式：

  1、consumer.receive() 或 consumer.receive(int timeout)；

  2、注册一个MessageListener。

  采用第一种方式，消息的接收者会一直等待下去，直到有消息到达，或者超时。后一种方式会注册一个监听器，当有消息到达的时候，会回调它的onMessage()方法。下面举例说明：

 MessageConsumer comsumer = session.createConsumer(queue);

 comsumer.setMessageListener(new MessageListener(){

            @Override

            public void onMessage(Message m) {

                TextMessage textMsg = (TextMessage) m;

                try {

                    System.out.println(textMsg.getText());

                } catch (JMSException e) {

                    e.printStackTrace();

                }

            }

           
        });

实战Queue

Queue实现的是点到点模型，在下面的例子中，启动2个消费者共同监听一个Queue，然后循环给这个Queue中发送多个消息，我们依然采用ActiveMQ。

import javax.jms.Connection;

import javax.jms.DeliveryMode;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.MessageProducer;

import javax.jms.Queue;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

import org.apache.activemq.command.ActiveMQQueue;

public class QueueTest {

    public static void main(String[] args) throws Exception {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");

   

        Connection connection = factory.createConnection();

        connection.start();

       

        //创建一个Queue

        Queue queue = new ActiveMQQueue("testQueue");

        //创建一个Session

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

       

        //注册消费者1

        MessageConsumer comsumer1 = session.createConsumer(queue);

        comsumer1.setMessageListener(new MessageListener(){

            public void onMessage(Message m) {

                try {

                    System.out.println("Consumer1 get " + ((TextMessage)m).getText());

                } catch (JMSException e) {

                    e.printStackTrace();

                }

            }

        });

       

        //注册消费者2

        MessageConsumer comsumer2 = session.createConsumer(queue);

        comsumer2.setMessageListener(new MessageListener(){

            public void onMessage(Message m) {

                try {

                    System.out.println("Consumer2 get " + ((TextMessage)m).getText());

                } catch (JMSException e) {

                    e.printStackTrace();

                }

            }

           

        });

       

        //创建一个生产者，然后发送多个消息。

        MessageProducer producer = session.createProducer(queue);

        for(int i=0; i<10; i++){

            producer.send(session.createTextMessage("Message:" + i));

        }

    }

}

  运行这个例子会得到下面的输出结果：

Consumer1 get Message:0

Consumer2 get Message:1

Consumer1 get Message:2

Consumer2 get Message:3

Consumer1 get Message:4

Consumer2 get Message:5

Consumer1 get Message:6

Consumer2 get Message:7

Consumer1 get Message:8

Consumer2 get Message:9

  可以看出每个消息直被消费了一次，但是如果有多个消费者同时监听一个Queue的话，无法确定一个消息最终会被哪一个消费者消费。

实战Topic

与Queue不同的是，Topic实现的是发布/订阅模型，在下面的例子中，启动2个消费者共同监听一个Topic，然后循环给这个Topic中发送多个消息。

import javax.jms.Connection;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import org.apache.activemq.command.ActiveMQTopic;

public class TopicTest {

    public static void main(String[] args) throws Exception {

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");

   

        Connection connection = factory.createConnection();

        connection.start();

       

        //创建一个Topic

        Topic topic= new ActiveMQTopic("testTopic");

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        

        //注册消费者1

        MessageConsumer comsumer1 = session.createConsumer(topic);

        comsumer1.setMessageListener(new MessageListener(){

            public void onMessage(Message m) {

                try {

                    System.out.println("Consumer1 get " + ((TextMessage)m).getText());

                } catch (JMSException e) {

                    e.printStackTrace();

                }

            }

        });

       

        //注册消费者2

        MessageConsumer comsumer2 = session.createConsumer(topic);

        comsumer2.setMessageListener(new MessageListener(){

            public void onMessage(Message m) {

                try {

                    System.out.println("Consumer2 get " + ((TextMessage)m).getText());

                } catch (JMSException e) {

                    e.printStackTrace();

                }

            }

           

        });

       

        //创建一个生产者，然后发送多个消息。

        MessageProducer producer = session.createProducer(topic);

        for(int i=0; i<10; i++){

            producer.send(session.createTextMessage("Message:" + i));

        }

    }

}

运行后得到下面的输出结果：

Consumer1 get Message:0

Consumer2 get Message:0

Consumer1 get Message:1

Consumer2 get Message:1

Consumer1 get Message:2

Consumer2 get Message:2

Consumer1 get Message:3

Consumer2 get Message:3

Consumer1 get Message:4

Consumer2 get Message:4

Consumer1 get Message:5

Consumer2 get Message:5

Consumer1 get Message:6

Consumer2 get Message:6

Consumer1 get Message:7

Consumer2 get Message:7

Consumer1 get Message:8

Consumer2 get Message:8

Consumer1 get Message:9

Consumer2 get Message:9

说明每一个消息都会被所有的消费者消费。
消息头

一个消息对象分为三部分：消息头(Headers)，属性（Properties）和消息体（Payload）。对于StreamMessage和MapMessage，消息本身就有特定的结构，而对于TextMessage，ObjectMessage和BytesMessage是无结构的。一个消息可以包含一些重要的数据或者仅仅是一个事件的通知。

    消息的Headers部分通常包含一些消息的描述信息，它们都是标准的描述信息。包含下面一些值：

　　JMSDestination

       消息的目的地，Topic或者是Queue。

　　JMSDeliveryMode

        消息的发送模式：persistent或nonpersistent。前者表示消息在被消费之前，如果JMS提供者DOWN了，重新启动后消息仍然存在。后者在这种情况下表示消息会被丢失。可以通过下面的方式设置：

       Producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

       JMSTimestamp

       当调用send()方法的时候，JMSTimestamp会被自动设置为当前事件。可以通过下面方式得到这个值：

       long timestamp = message.getJMSTimestamp();

　　JMSExpiration

       表示一个消息的有效期。只有在这个有效期内，消息消费者才可以消费这个消息。默认值为0，表示消息永不过期。可以通过下面的方式设置：

       producer.setTimeToLive(3600000); //有效期1小时 （1000毫秒 * 60秒 * 60分）

　　JMSPriority

       消息的优先级。0-4为正常的优先级，5-9为高优先级。可以通过下面方式设置：

       producer.setPriority(9);

　　JMSMessageID

       一个字符串用来唯一标示一个消息。

　　JMSReplyTo

       有时消息生产者希望消费者回复一个消息，JMSReplyTo为一个Destination，表示需要回复的目的地。当然消费者可以不理会它。

　　JMSCorrelationID

       通常用来关联多个Message。例如需要回复一个消息，可以把JMSCorrelationID设置为所收到的消息的JMSMessageID。

　　JMSType

       表示消息体的结构，和JMS提供者有关。

　　JMSRedelivered

       如果这个值为true，表示消息是被重新发送了。因为有时消费者没有确认他已经收到消息或者JMS提供者不确定消费者是否已经收到。

    除了Header，消息发送者可以添加一些属性(Properties)。这些属性可以是应用自定义的属性，JMS定义的属性和JMS提供者定义的属性。我们通常只适用自定义的属性。

    后面会讲到这些Header和属性的用法。