Q來作為jms匯流排，並且給大家介紹了activeMQ的叢集和高可用部署方案，本期給大家再介紹下，如何根據自己的專案需求，更好地使用activeMQ的兩種訊息處理模式。

1    queue與topic的技術特點對比

          Topic和queue的最大區別在於topic是以廣播的形式，通知所有線上監聽的客戶端有新的訊息，沒有監聽的客戶端將收不到訊息；而queue則是以點對點的形式通知多個處於監聽狀態的客戶端中的一個。

2    topic和queue方式的訊息處理效率比較

        通過增加監聽客戶端的併發數來驗證，topic的訊息推送，是否會因為監聽客戶端的併發上升而出現明顯的下降，測試環境的伺服器為ci環境的ActiveMQ，客戶端為我的本機。

        從實測的結果來看，topic方式傳送的訊息，傳送和接收的效率，在一個訂閱者和100個訂閱者的前提下沒有明顯差異，但在500個訂閱者（執行緒）併發的前提下，效率差異很明顯（由於500執行緒併發的情況下，我本機的cpu佔用率已高達70-90%，所以無法確認是我本機測試造成的效能瓶頸還是topic訊息傳送方式存在效能瓶頸，造成效率下降如此明顯）。

        Topic方式傳送的訊息與queue方式傳送的訊息，傳送和接收的效率，在一個訂閱者和100個訂閱者的前提下沒有明顯差異，但在500個訂閱者併發的前提下，topic方式的效率明顯低於queue。

        Queue方式傳送的訊息，在一個訂閱者、100個訂閱者和500個訂閱者的前提下，傳送和接收的效率沒有明顯變化。

Topic實測資料：

Queue實測資料：

3     topic方式的訊息處理示例

3.1     通過客戶端程式碼呼叫來發送一個topic的訊息：

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.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

publicclass SendTopic {

    privatestaticfinalint SEND_NUMBER = 5;

    publicstaticvoid sendMessage(Session session, MessageProducer producer)

            throws Exception {

         for ( int i = 1; i <= SEND_NUMBER; i++) {

            TextMessage message = session

                    .createTextMessage("ActiveMq傳送的訊息" + i);

            //傳送訊息到目的地方

            System. out.println("傳送訊息：" + "ActiveMq 傳送的訊息" + i);

            producer.send(message);

        }

    }

    publicstaticvoid main(String[] args) {

        // ConnectionFactory：連線工廠，JMS用它建立連線

        ConnectionFactory connectionFactory;

        // Connection：JMS客戶端到JMS Provider的連線

        Connection connection = null;

        // Session：一個傳送或接收訊息的執行緒

        Session session;

        // Destination：訊息的目的地;訊息傳送給誰.

        Destination destination;

        // MessageProducer：訊息傳送者

        MessageProducer producer;

        // TextMessage message;

        //構造ConnectionFactory例項物件，此處採用ActiveMq的實現jar

        connectionFactory = new ActiveMQConnectionFactory(

                ActiveMQConnection. DEFAULT_USER,

                ActiveMQConnection. DEFAULT_PASSWORD,

                "tcp://10.20.8.198:61616");

        try {

            //構造從工廠得到連線物件

            connection = connectionFactory.createConnection();

            //啟動

            connection.start();

            //獲取操作連線

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

            //獲取session注意引數值FirstTopic是一個伺服器的topic(與queue訊息的傳送相比，這裡是唯一的不同)

            destination = session.createTopic("FirstTopic");

            //得到訊息生成者【傳送者】

            producer = session.createProducer(destination);

            //設定不持久化，此處學習，實際根據專案決定

            producer.setDeliveryMode(DeliveryMode. PERSISTENT);

            //構造訊息，此處寫死，專案就是引數，或者方法獲取

            sendMessage(session, producer);

            session.commit();

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            try {

                if ( null != connection)

                    connection.close();

            } catch (Throwable ignore) {

            }

        }

    }

}

3.2     啟動多個客戶端監聽來接收topic的訊息：

publicclass ReceiveTopic implements Runnable {

      private StringthreadName;

      ReceiveTopic(String threadName) {

           this.threadName = threadName;

      }

      publicvoid run() {

           // ConnectionFactory：連線工廠，JMS用它建立連線

           ConnectionFactory connectionFactory;

           // Connection：JMS客戶端到JMS Provider的連線

           Connection connection = null;

           // Session：一個傳送或接收訊息的執行緒

           Session session;

           // Destination：訊息的目的地;訊息傳送給誰.

           Destination destination;

           //消費者，訊息接收者

           MessageConsumer consumer;

           connectionFactory = new ActiveMQConnectionFactory(

                      ActiveMQConnection. DEFAULT_USER,

                      ActiveMQConnection. DEFAULT_PASSWORD,"tcp://10.20.8.198:61616");

           try {

                 //構造從工廠得到連線物件

                 connection = connectionFactory.createConnection();

                 //啟動

                 connection.start();

                 //獲取操作連線,預設自動向伺服器傳送接收成功的響應

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

                 //獲取session注意引數值FirstTopic是一個伺服器的topic

                 destination = session.createTopic("FirstTopic");

                 consumer = session.createConsumer(destination);

                 while ( true) {

                      //設定接收者接收訊息的時間，為了便於測試，這裡設定為100s

                      TextMessage message = (TextMessage) consumer

                                  .receive(100 * 1000);

                      if ( null != message) {

                            System. out.println("執行緒"+threadName+"收到訊息:" + message.getText());

                      } else {

                            continue;

                      }

                 }

           } catch (Exception e) {

                 e.printStackTrace();

           } finally {

                 try {

                      if ( null != connection)

                            connection.close();

                 } catch (Throwable ignore) {

                 }

           }

      }

      publicstaticvoid main(String[] args) {

            //這裡啟動3個執行緒來監聽FirstTopic的訊息，與queue的方式不一樣三個執行緒都能收到同樣的訊息

           ReceiveTopic receive1= new ReceiveTopic("thread1");

           ReceiveTopic receive2= new ReceiveTopic("thread2");

           ReceiveTopic receive3= new ReceiveTopic("thread3");

           Thread thread1= new Thread(receive1);

           Thread thread2= new Thread(receive2);

           Thread thread3= new Thread(receive3);

           thread1.start();

           thread2.start();

           thread3.start();

      }

}

4     queue方式的訊息處理示例

         參考上一期文章：開源jms服務ActiveMQ的負載均衡+高可用部署方案探索。