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"); 27         //創建JMS producer
        MessageProducer producer = session.createProducer(destination); 29         
        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"); 28         //創建JMS consumer
        MessageConsumer consumer = session.createConsumer(destination); 30         
        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"); 25         //創建JMS producer
        MessageProducer producer = session.createProducer(destination); 27  producer.setDeliveryMode(DeliveryMode.PERSISTENT); 28         //啟動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"); 27         //創建JMS consumer
        TopicSubscriber ts = session.createDurableSubscriber(destination, "TT"); 29         //啟動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的默認傳送模式,此模式保證這些消息只被傳送一次和成功使用一次。對於這些消息，可靠性是優先考慮的因素。可靠性的另一個重要方面是確保持久性消息傳送至目標后，消息服務在向消費者傳送它們之前不會丟失這些消息。

　　這意味着在持久性消息傳送至目標時，消息服務將其放入持久性數據存儲。如果消息服務由於某種原因導致失敗，它可以恢復此消息並將此消息傳送至相應的消費者。雖然這樣增加了消息傳送的開銷，但是卻增加了可靠性。

非持久化消息

　　保證這些消息最多被傳送一次。對於這些消息，可靠性並非主要的考慮因素。此模式並不要求持久性的數據存儲，也不保證消息服務由於某種原因導致失敗后消息不會丟失。