JMS一些实例六到实例十

继续上篇文章jms实例六到十

 

深入掌握JMS（六）：消息头

一个消息对象分为三部分：消息头(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和属性的用法。

深入掌握JMS（七）：DeliveryMode例子

在下面的例子中，分别发送一个Persistent和nonpersistent的消息，然后关闭退出JMS。

            import javax.jms.Connection;
            import javax.jms.DeliveryMode;
            import javax.jms.MessageProducer;
            import javax.jms.Queue;
            import javax.jms.Session;
            import org.apache.activemq.ActiveMQConnectionFactory;
            import org.apache.activemq.command.ActiveMQQueue;

            public class DeliveryModeSendTest {

                public static void main(String[] args) throws Exception {
                    ActiveMQConnectionFactory factory = new
            ActiveMQConnectionFactory("vm://localhost");
                    Connection connection = factory.createConnection();
                    connection.start();
                    Queue queue = new ActiveMQQueue("testQueue");
                    Session session = connection.createSession(false,
            Session.AUTO_ACKNOWLEDGE);
                    MessageProducer producer = session.createProducer(queue);
                    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                    producer.send(session.createTextMessage("A persistent
            Message"));
                    producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                    producer.send(session.createTextMessage("A non persistent
            Message"));
                    System.out.println("Send messages sucessfully!");
                }
            }

                运行上面的程序，当输出“Send messages
            sucessfully!”时，说明两个消息都已经发送成功，然后我们结束它，来停止JMS Provider。

                接下来我们重新启动JMS Provicer，然后添加一个消费者：

            import javax.jms.Connection;
            import javax.jms.JMSException;
            import javax.jms.Message;
            import javax.jms.MessageConsumer;
            import javax.jms.MessageListener;
            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 DeliveryModeReceiveTest {

                public static void main(String[] args) throws Exception {
                    ActiveMQConnectionFactory factory = new
            ActiveMQConnectionFactory("vm://localhost");
                    Connection connection = factory.createConnection();
                    connection.start();
                    Queue queue = new ActiveMQQueue("testQueue");
                    Session session = connection.createSession(false,
            Session.AUTO_ACKNOWLEDGE);
                    MessageConsumer comsumer = session.createConsumer(queue);
                    comsumer.setMessageListener(new MessageListener(){
                        public void onMessage(Message m) {
                            try {
                                System.out.println("Consumer get " +
            ((TextMessage)m).getText());
                            } catch (JMSException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
            }

            运行上面的程序，可以得到下面的输出结果：

            Consumer get A persistent Message

            可以看出消息消费者只接收到一个消息，它是一个Persistent的消息。

            而刚才发送的non persistent消息已经丢失了。

            另外, 如果发送一个non persistent消息, 而刚好这个时候没有消费者在监听, 这个消息也会丢失.

深入掌握JMS（八）：JMSReplyTo

在下面的例子中，首先创建两个Queue，发送者给一个Queue发送，

接收者接收到消息之后给另一个Queue回复一个Message，然后再创建一个消费者来接受所回复的消息。

 

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.Queue; 
            import javax.jms.Session; 
            import javax.jms.TextMessage; 

            import org.apache.activemq.ActiveMQConnectionFactory; 
            import org.apache.activemq.command.ActiveMQQueue; 

            public class MessageSendReceiveAndReply { 

                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"); 

                    //消息回复到这个Queue 
                    Queue replyQueue = new ActiveMQQueue("replyQueue"); 
                    final Session session = connection.createSession(false, 
            Session.AUTO_ACKNOWLEDGE); 

                    //创建一个消息，并设置它的JMSReplyTo为replyQueue。 
                    Message message = session.createTextMessage("Andy"); 
                    message.setJMSReplyTo(replyQueue); 
                    MessageProducer producer = session.createProducer(queue); 
                    producer.send(message); 
                    //消息的接收者 
                    MessageConsumer comsumer = session.createConsumer(queue); 
                    comsumer.setMessageListener(new MessageListener(){ 
                        public void onMessage(Message m) { 
                            try { 
                                //创建一个新的MessageProducer来发送一个回复消息。 
                                MessageProducer producer = 
            session.createProducer(m.getJMSReplyTo()); 
                                producer.send(session.createTextMessage("Hello " 
            + ((TextMessage) m).getText())); 
                            } catch (JMSException e1) { 
                                e1.printStackTrace(); 
                            } 
                        } 
                    }); 
                    //这个接收者用来接收回复的消息 
                    MessageConsumer comsumer2 = 
            session.createConsumer(replyQueue); 
                    comsumer2.setMessageListener(new MessageListener(){ 
                        public void onMessage(Message m) { 
                            try { 
                                System.out.println(((TextMessage) m).getText()); 
                            } catch (JMSException e) { 
                                e.printStackTrace(); 
                            } 
                        } 
                    }); 
                } 

            } 

                首先消息生产者发送一个消息，内容为“Andy”， 
            然后消费者收到这个消息之后根据消息的JMSReplyTo，回复一个消息，内容为“Hello Andy'。 
            最后在回复的Queue上创建一个接收回复消息的消费者，它输出所回复的内容。 

                运行上面的程序，可以得到下面的输出结果： 

            Hello Andy

 

深入掌握JMS（九）：Selector

前面的例子中创建一个消息消费者使用的是:
                      sesssion.createConsumer(destination)
              另外，还提供了另一种方式：
                sesssion.createConsumer(destination, selector)
            这里selector是一个字符串，用来过滤消息。也就是说，这种方式可以创建一个可以只接收特定消息的一个消费者。Selector的格式是类似于SQL-92的一种语法。可以用来比较消息头信息和属性。

              下面的例子中，创建两个消费者，共同监听同一个Queue，但是它们的Selector不同，然后创建一个消息生产者，来发送多个消息。

 

 

       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.Queue; 
            import javax.jms.Session; 
            import javax.jms.TextMessage; 

            import org.apache.activemq.ActiveMQConnectionFactory; 
            import org.apache.activemq.command.ActiveMQQueue; 

            public class JMSSelectorTest { 

                public static void main(String[] args) throws Exception { 
                    ActiveMQConnectionFactory factory = new 
            ActiveMQConnectionFactory("vm://localhost"); 
                    Connection connection = factory.createConnection(); 
                    connection.start(); 
                    Queue queue = new ActiveMQQueue("testQueue"); 
                    Session session = connection.createSession(false, 
            Session.AUTO_ACKNOWLEDGE); 
                    MessageConsumer comsumerA = session.createConsumer(queue, 
            "receiver = 'A'"); 
                    comsumerA.setMessageListener(new MessageListener(){ 
                        public void onMessage(Message m) { 
                            try { 
                                System.out.println("ConsumerA get " + 
            ((TextMessage) m).getText()); 
                            } catch (JMSException e1) { } 
                        } 
                    }); 
                    MessageConsumer comsumerB = session.createConsumer(queue, 
            "receiver = 'B'"); 
                    comsumerB.setMessageListener(new MessageListener(){ 
                        public void onMessage(Message m) { 
                            try { 
                                System.out.println("ConsumerB get " + 
            ((TextMessage) m).getText()); 
                            } catch (JMSException e) { } 
                        } 
                    }); 
                    MessageProducer producer = session.createProducer(queue); 
                    for(int i=0; i<10; i++) { 
                        String receiver = (i%3 == 0 ? "A" : "B"); 
                        TextMessage message = 
            session.createTextMessage("Message" + i + ", receiver:" + receiver); 
                        message.setStringProperty("receiver", receiver); 
                        producer.send(message ); 
                    } 
                } 
            } 

            结果如下： 
            ConsumerA get Message0, receiver:A 
            ConsumerB get Message1, receiver:B 
            ConsumerB get Message2, receiver:B 
            ConsumerA get Message3, receiver:A 
            ConsumerB get Message4, receiver:B 
            ConsumerB get Message5, receiver:B 
            ConsumerA get Message6, receiver:A 
            ConsumerB get Message7, receiver:B 
            ConsumerB get Message8, receiver:B 
            ConsumerA get Message9, receiver:A 

            可以看出，消息消费者只会取走它自己感兴趣的消息。

 

 

深入掌握JMS（十）：JMSCorrelationID与Selector

前面讲过JMSCorrelationID主要是用来关联多个Message，例如需要回复一个消息的时候，通常把回复的消息的
           JMSCorrelationID设置为原来消息的ID。在下面这个例子中，创建了三个消息生产者A，B，C和三个消息消费者A，B，C。生产者A给消费者A发送一个消息，同时需要消费者A给它回复一个消息。B、C与A类似。
               简图如下：
                  生产者A-----发送----〉消费者A-----回复------〉生产者A
                   生产者B-----发送----〉消费者B-----回复------〉生产者B
                   生产者C-----发送----〉消费者C-----回复------〉生产者C
               需要注意的是，所有的发送和回复都使用同一个Queue，通过Selector区分。

 

       import javax.jms.*; 
           import org.apache.activemq.ActiveMQConnectionFactory; 
           import org.apache.activemq.command.ActiveMQQueue; 

           public class JMSCorrelationIDTest { 
               private Queue queue; 
               private Session session; 
               public JMSCorrelationIDTest() throws JMSException{ 
                   ActiveMQConnectionFactory factory = new 
           ActiveMQConnectionFactory("vm://localhost"); 
                   Connection connection = factory.createConnection(); 
                   connection.start(); 
                   queue = new ActiveMQQueue("testQueue"); 
                   session = connection.createSession(false, 
           Session.AUTO_ACKNOWLEDGE); 
                   setupConsumer("ConsumerA"); 
                   setupConsumer("ConsumerB"); 
                   setupConsumer("ConsumerC"); 
                   setupProducer("ProducerA", "ConsumerA"); 
                   setupProducer("ProducerB", "ConsumerB"); 
                   setupProducer("ProducerC", "ConsumerC"); 
               } 
               private void setupConsumer(final String name) throws 
           JMSException { 
                   //创建一个消费者，它只接受属于它自己的消息 
                   MessageConsumer consumer = session.createConsumer(queue, 
           "receiver='" + name + "'"); 
                   consumer.setMessageListener(new MessageListener(){ 
                       public void onMessage(Message m) { 
                           try { 
                               MessageProducer producer = 
           session.createProducer(queue); 
                               System.out.println(name + " get:" + 
           ((TextMessage)m).getText()); 
                               //回复一个消息 
                               Message replyMessage = 
           session.createTextMessage("Reply from " + name); 
                               //设置JMSCorrelationID为刚才收到的消息的ID 
           replyMessage.setJMSCorrelationID(m.getJMSMessageID()); 
                               producer.send(replyMessage); 
                           } catch (JMSException e) { } 
                       } 
                   }); 
               } 

               private void setupProducer(final String name, String 
           consumerName) throws JMSException { 
                   MessageProducer producer = session.createProducer(queue); 
                   producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); 
                   //创建一个消息，并设置一个属性receiver，为消费者的名字。 
                   Message message = session.createTextMessage("Message from " 
           + name); 
                   message.setStringProperty("receiver", consumerName); 
                   producer.send(message); 
                   //等待回复的消息 
                   MessageConsumer replyConsumer = 
           session.createConsumer(queue, "JMSCorrelationID='" + 
           message.getJMSMessageID() + "'"); 
                   replyConsumer.setMessageListener(new MessageListener(){ 
                       public void onMessage(Message m) { 
                           try { 
                               System.out.println(name + " get reply:" + 
           ((TextMessage)m).getText()); 
                           } catch (JMSException e) { } 
                       } 
                   }); 
               } 
               public static void main(String[] args) throws Exception { 
                   new JMSCorrelationIDTest (); 
               } 
           } 

           运行结果为： 
           ConsumerA get:Message from ProducerA 
           ProducerA get reply:Reply from ConsumerA 
           ConsumerB get:Message from ProducerB 
           ProducerB get reply:Reply from ConsumerB 
           ConsumerC get:Message from ProducerC 
           ProducerC get reply:Reply from ConsumerC