在上一篇《ActiveMQ入門系列一：認識并安裝ActiveMQ（Windows下）》中，大致介紹了ActiveMQ和一些概念，并下載、安裝、啟動他，還訪問了他的控制臺頁面。

這篇，就用代碼實例說下如何實現消息的生產和消費。

一、理論基礎

同RabbitMQ一樣，ActiveMQ中也是有兩種模式：

• 點對點模式（Point to Point，簡寫為PTP）
• 發布/訂閱模式（Publish & Subscribe，簡寫為Pub & Sub）

通過上一篇我們知道了制造消息的應用叫生產者（Producer），生產者在生產了消息后會發送消息到目的地（Destination），到達消費和處理消息的應用（也就是消費者Consumer）。這里的兩種模式就通過對應不同的消息目的地（Destination）來實現，PTP對應Queue（隊列）、Pub&Sub對應Topic（主題）。

今天就詳細介紹下PTP和Queue，下一篇介紹Pub & Sub和Topic。

在PTP模式的示意圖：

?

• 消息生產者生產消息發送到queue中，然后消息消費者從queue中取出并且消費消息。
• 消息被消費以后，queue中不再有存儲，所以消息消費者不可能消費到已經被消費的消息。
• Queue支持存在多個消費者，但是對一個消息而言，只會有一個消費者可以消費、其它的則不能消費此消息了。
• 當消費者不存在時，消息會一直保存，直到有消費消費。

在PTP中，代碼實現有兩種方式：消費者主動消費和消費者監聽消費，下面就分別說下。

二、消費者主動消費

主動消費是最基本也是最簡單的消費方式，先上代碼：

創建maven工程并引入依賴 <dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-core</artifactId><version>5.7.0</version></dependency>

實現生產者 package com.sam.ptp;
import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; /*** @author JAVA開發老菜鳥**/ public class Producer {public static final String QUEUE_NAME = "ptp-demo";//隊列名public void producer(String message) throws JMSException {ConnectionFactory factory = null;Connection connection = null;Session session = null;MessageProducer producer = null;try {/*** 1.創建連接工廠* 創建工廠，構造方法有三個參數：分別是用戶名、密碼、連接地址* 無參構造：有默認的連接地址，localhost* 一個參數：無驗證模式，無用戶的認證* 三個參數：有認證和連接地址，我這里使用三個參數的構造方法*/factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");/*** 2.創建連接，有兩個方法（我這里使用無參數的）* 無參數* 有參數：用戶名、密碼；*/connection = factory.createConnection();/*** 3.啟動連接* 生產者可以不用調用start()方法啟動，因為在發送消息的時候回進行檢查* 如果未啟動連接，會自動啟動。* 如果有特殊配置，需要配置完成后再啟動連接*/connection.start();/*** 4.用連接創建會話* 有兩個參數：是否需要事務、消息確認機制* 如果支持事務，對于生產者來說第二個參數就無效了，這個時候第二個參數建議傳入Session.SESSION_TRANSACTED* 如果不支持事務，第二個參數有效且必須傳遞** AUTO_ACKNOWLEDGE：自動確認，消息處理后自動確認（商業開發不推薦）* CLIENT_ACKNOWLEDGE：客戶端手動確認，消費者處理后必須手動確認* DUPS_OK_ACKNOWLEDGE：有副本的客戶端手動確認，消息可以多次處理（不建議）*/session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);/*** 5.用會話創建目的地（隊列）、生產者、消息* 隊列名是隊列的唯一標記* 創建生產者的時候可以指定目的地，也可以在發送消息的時候再指定*/Destination destination = session.createQueue(QUEUE_NAME);producer = session.createProducer(destination);TextMessage textMessage = session.createTextMessage(message);/*** 6.生產者發送消息到目的地*/producer.send(textMessage);System.out.println("消息發送成功");} catch(Exception ex){throw ex;} finally {/*** 7.釋放資源*/if(producer != null){producer.close();}if(session != null){session.close();}if(connection != null){connection.close();}}}public static void main(String[] args){Producer producer = new Producer();try{producer.producer("hello, activemq");} catch (Exception ex){ex.printStackTrace();}} }

?

實現消費者 package com.sam.ptp;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;/*** @author JAVA開發老菜鳥** 主動消費*/ public class Consumer {public String consumer() throws JMSException {ConnectionFactory factory = null;Connection connection = null;Session session = null;MessageConsumer consumer = null;try {factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");connection = factory.createConnection();/*** 消費者必須啟動連接，否則無法消費*/connection.start();session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination destination = session.createQueue(Producer.QUEUE_NAME);consumer = session.createConsumer(destination);/*** 獲取隊列消息*/Message message = consumer.receive();String text = ((TextMessage) message).getText();return text;} catch(Exception ex){throw ex;} finally {/*** 7.釋放資源*/if(consumer != null){consumer.close();}if(session != null){session.close();}if(connection != null){connection.close();}}}public static void main(String[] args){Consumer consumer = new Consumer();try{String message = consumer.consumer();System.out.println("消息消費成功：" + message);} catch (Exception ex){ex.printStackTrace();}} }

?

好，這樣代碼就寫好了，我們來測試下。

1.先運行生產者，我發現報錯了。。。

好吧，原來是我這次沒有啟動ActiveMQ，被自己蠢哭了。。。

啟動ActiveMQ之后，再運行生產者，成功了。

去看下控制臺頁面的變化，隊列里面多了個“ptp-demo”隊列，這個就是我們生產者代碼里面的隊列名，并且能看到該隊列的基本情況：

從左到右依次為，有待消費的消息1條、消費者0個、已經發送的消息1條、已經消費的消息0條

2.接下來運行消費者，成功

再去看下控制臺頁面，發現隊列信息變了，從左到右依次為：有待消費的消息0條、消費者0個、已經發送的消息1條、已經消費的消息1條

也就是說，消息真的被消費了！

?代碼寫完了，也按照預期執行完了，我們現在再回過頭來分析下消費者的代碼，會發現他在consumer.receive()之后不會再消費其他消息了，即便后面再有消息被生產出來也不會再消費。也就是說只能在運行后消費一次消息，這個就是主動消費。

如果想要循環消費多次產生的消息的話，怎么辦呢？請用下面的監聽消費

?

三、消費者監聽消費

還是先上代碼，代碼結構同主動消費類似，有細微差別，具體代碼不貼了，可以到我的GitHub或碼云上獲取源碼

首先為了區分，我把隊列名改了 public static final String QUEUE_NAME = "ptp-listener-demo";//隊列名

?

生產者和消費者的消息確認方式都改成了客戶端手動確認，不再自動確認，手動確認有個好處就是可以防止消息沒有被正常消費而丟失，這個同RabbitMQ機制一樣 session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

?

生產者生產消息的時候，為了方便我改成了一次性發送10條 /*** 6.創建消息并且生產者發送消息到目的地*/for(int num = 0; num < 10; num++){TextMessage textMessage = session.createTextMessage(message + num);producer.send(textMessage);System.out.println("消息發送成功"+textMessage.getText());}

?

關鍵點來了，在消費者上加了一個監聽器 /*** 注冊監聽器，隊列中的消息變化會自動觸發監聽器，接收并自動處理消息** 監聽器一旦注冊，永久有效，一直到程序關閉* 監聽器可以注冊多個，相當于集群* activemq自動輪詢多個監聽器，實現并行處理*/consumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {try {//需要手動確認消息 message.acknowledge();TextMessage om = (TextMessage) message;String data = om.getText();System.out.println(data);} catch (JMSException e) {e.printStackTrace();}}});

?

執行生產者：

?

執行消費者，消息全部被消費了：

?

再執行2遍生產者，消息同樣都被消費了。?

控制臺頁面多了個隊列，由于監聽中的消費者沒有關閉，因此這里能看到消費者數量為1，我執行了三遍生產者，因此消息有30條。

還沒完，繼續...

我們這次先啟動2個消費者，然后啟動生產者

兩個生產者分別消費了消息0,2,4,6,8和1,3,5,7,9

也就是說兩個消費者都監聽到了消息，并且activemq自動輪詢兩個監聽器發送消息。

?

好，到這里，ActiveMQ的點對點模式就介紹完了。下一篇介紹發布訂閱模式，敬請期待

轉載于:https://www.cnblogs.com/sam-uncle/p/10988930.html

總結

以上是生活随笔為你收集整理的ActiveMQ入门系列二：入门代码实例（点对点模式）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。