JMS开发步骤和持久化/非持久化Topic消息

------------------------------------------------

开发一个JMS的基本步骤如下:

　　1.创建一个JMS connection factory

　　2.通过connection factory来创建JMS connection

　　3.启动JMS connection

　　4.通过connection创建JMS session

　　5.创建JMS destination

　　6.创建JMS producer 或者创建JMS message,并设置destination

　　7.创建JMS consumer 或者注册一个JMS message listener

　　8.发送或者接受JMS message

　　9.关闭所有的JMS资源(connection、session、producer、consumer等)

可以参考下图：

非持久的Topic消息示例

　　对于非持久化的消息，当发送方发送消息的时候:

　　　　如果接收方不在线，则接收方永远也收不到这些消息了

　　　　如果接收方在线，则接收方会收到这些消息

1、消息发送程序

 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnectionFactory;

 /**
  * 非持久化Topic消息发送者
  * @author Administrator
  *
  */
 public class NoPersistenceSender {
     public static void main(String[] args) throws Exception {
         //创建一个JMS connection factory
         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
         //通过connection factory来创建JMS connection
         Connection connection = connectionFactory.createConnection();
         //启动JMS connection
         connection.start();
         //通过connection创建JMS session
         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
         //创建JMS destination
Destination destination = session.createTopic("noPersistenceTopic");
         //创建JMS producer
MessageProducer producer = session.createProducer(destination);

         for(int i = 0;i < 10;i++){
             TextMessage message = session.createTextMessage("message-"+i);
             //发送message
             producer.send(message);
         }
         //关闭所有的JMS资源
         session.commit();
         session.close();
         connection.close();
     }
 }

运行完消息发送程序后，可以访问192.168.1.81:8161

2、消息接收程序

　　对于非持久的Topic消息的接收需要注意以下几点:

　　　　a.接收程序必须在线，然后消息发送方再发送消息，接收程序才能接收到消息

　　　　b.由于不知道消息发送方要发送多少条消息，所以利用while循环的方式来接收消息

　　　　c.如果接收程序不在线，此时发送程序发送了消息的话，则该消息将永远不会被接收方收到。

 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.Session;
 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnectionFactory;

 /**
  * 非持久化Topic消息接收者
  * @author Administrator
  *
  */
 public class NoPersistenceReceiver {
     public static void main(String[] args) throws Exception {
         //创建一个JMS connection factory
         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
         //通过connection factory来创建JMS connection
         Connection connection = connectionFactory.createConnection();
         //启动JMS connection
         connection.start();
         //通过connection创建JMS session
         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
         //创建JMS destination
Destination destination = session.createTopic("noPersistenceTopic");
         //创建JMS consumer
MessageConsumer consumer = session.createConsumer(destination);

         Message message = consumer.receive();
         while(message != null){
             TextMessage txtMsg = (TextMessage)message;
             System.out.println("收到消息:"+txtMsg.getText());
             message = consumer.receive();
         }
         //关闭所有的JMS资源
         session.commit();
         session.close();
         connection.close();
     }
 }

运行结果:

持久的Topic消息示例

1.消息发送程序

　　对于持久的Topic消息的发送方需要注意以下几点:

　　　　a.要用持久化订阅，发送消息者要用DeliveryMode.PERSISTENT模式来发送

 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;

 import org.apache.activemq.ActiveMQConnectionFactory;

 /**
  * 持久化Topic消息发送者
  * @author Administrator
  */
 public class PersistenceSender {
     public static void main(String[] args) throws Exception {
         //创建一个JMS connection factory
         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
         //通过connection factory来创建JMS connection
         Connection connection = connectionFactory.createConnection();
         //通过connection创建JMS session
         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
         //创建JMS destination
Destination destination = session.createTopic("PersistenceTopic");
         //创建JMS producer
MessageProducer producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
         //启动JMS connection
         connection.start();
         for(int i = 0;i < 10;i++){
             TextMessage message = session.createTextMessage("message-"+i);
             //发送message
             producer.send(message);
         }
         //关闭所有的JMS资源
         session.commit();
         session.close();
         connection.close();
     }
 }

2.消息接收程序

　　对于持久的Topic消息的接收方需要注意以下几点:

　　　　a.需要在连接上设置消费者id,用来识别消费者

　　　　b.需要创建TopicSubscriber来订阅

　　　　c.一定要先运行一次该消费者程序，等于向消费服务中间件注册这个消费者，然后再运行消息发送者来发送消息，这样的话，无论消费者是否在线都会收到消息，如果不在线的话，则下次连接的时候会把没有收过的消息都接收下来。

 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Message;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 import javax.jms.Topic;
 import javax.jms.TopicSubscriber;

 import org.apache.activemq.ActiveMQConnectionFactory;

 /**
  * 持久化Topic消息接收者
  * @author Administrator
  *
  */
 public class PersistenceReceiver {
     public static void main(String[] args) throws Exception {
         //创建一个JMS connection factory
         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://120.76.123.81:61616");
         //通过connection factory来创建JMS connection
         Connection connection = connectionFactory.createConnection();
         connection.setClientID("con1");
         //通过connection创建JMS session
         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
         //创建JMS destination
Topic destination = session.createTopic("PersistenceTopic");
         //创建JMS consumer
TopicSubscriber ts = session.createDurableSubscriber(destination, "TT");
         //启动JMS connection
         connection.start();
         Message message = ts.receive();
         while(message != null){
             TextMessage txtMsg = (TextMessage)message;
             session.commit();
             System.out.println("收到消息:"+txtMsg.getText());
             message = ts.receive(1000L);
         }
         //关闭所有的JMS资源
         session.close();
         connection.close();
     }
 }

关于持久化和非持久化消息

有两种方式指定传送模式:

　　1.使用setDeliveryMode方法，这样所有的消息都采用此传送模式;如producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)

　　2.使用send方法为每条消息设置传送模式

持久化消息

　　这是ActiveMQ的默认传送模式,此模式保证这些消息只被传送一次和成功使用一次。对于这些消息，可靠性是优先考虑的因素。可靠性的另一个重要方面是确保持久性消息传送至目标后，消息服务在向消费者传送它们之前不会丢失这些消息。

　　这意味着在持久性消息传送至目标时，消息服务将其放入持久性数据存储。如果消息服务由于某种原因导致失败，它可以恢复此消息并将此消息传送至相应的消费者。虽然这样增加了消息传送的开销，但是却增加了可靠性。

非持久化消息

　　保证这些消息最多被传送一次。对于这些消息，可靠性并非主要的考虑因素。此模式并不要求持久性的数据存储，也不保证消息服务由于某种原因导致失败后消息不会丢失。