在上一篇《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() { @Override public 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的點對點模式就介紹完了。下一篇介紹發布訂閱模式，敬請期待