深刻淺出 消息隊列 ActiveMQ------加強版
本小節咱們將講解Apache開源下的ActiveMQ，而ActiveMQ是JMS的一個具體實現。JMS即Java消息服務（Java Message Service）應用程序接口是一個Java平臺中關於面向消息中間件（MOM）的API，用於在兩個應用程序之間，或分佈式系統中發送消息，進行異步通訊。Java消息服務是一個與具體平臺無關的API，絕大多數MOM提供商都對JMS提供支持。html
在學習ActiveMQ以前，咱們有必要對JMS進行一下簡單的瞭解，因爲篇幅的限制，對於JMS的學習請參考下面的連接:java
JMS參考連接點擊這裏linux
1、 概述與介紹web
ActiveMQ 是Apache出品，最流行的、功能強大的即時通信和集成模式的開源服務器。ActiveMQ 是一個徹底支持JMS1.1和J2EE 1.4規範的 JMS Provider實現。提供客戶端支持跨語言和協議，帶有易於在充分支持JMS 1.1和1.4使用J2EE企業集成模式和許多先進的功能。spring
2、 特性apache
一、 多種語言和協議編寫客戶端。語言： Java、C、C++、C#、Ruby、Perl、Python、PHP。應用協議：OpenWire、Stomp REST、WS Notification、XMPP、AMQPwindows
二、徹底支持JMS1.1和J2EE 1.4規範 （持久化，XA消息，事務)服務器
三、對Spring的支持，ActiveMQ能夠很容易內嵌到使用Spring的系統裏面去，並且也支持Spring2.0的特性網絡
四、經過了常見J2EE服務器（如 Geronimo、JBoss 四、GlassFish、WebLogic)的測試，其中經過JCA 1.5 resource adaptors的配置，可讓ActiveMQ能夠自動的部署到任何兼容J2EE 1.4 商業服務器上session
五、支持多種傳送協議：in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA
六、支持經過JDBC和journal提供高速的消息持久化
七、從設計上保證了高性能的集羣，客戶端-服務器，點對點
八、支持Ajax
九、支持與Axis的整合
十、能夠很容易得調用內嵌JMS provider，進行測試
對於Active MQ更多的信息能夠點擊這裏
3、 安裝
開發環境：
System：Windows
JDK：1.6+
IDE：myeclipse
apache ActiveMQ 5.8
Email：1261245595@qq.com
Blog：http://www.cnblogs.com/xiohao
一、 下載ActiveMQ，下載地址：http://www.apache.org/dyn/closer.cgi?path=/activemq/apache-activemq/5.8.0/apache-activemq-5.8.0-bin.zip
二、 解壓apache-activemq-5.8.0.zip便可完成ActiveMQ的安裝
三、 解壓後目錄結構以下
+bin (windows下面的bat和unix/linux下面的sh) 啓動ActiveMQ的啓動服務就在這裏
+conf (activeMQ配置目錄，包含最基本的activeMQ配置文件）
+data （默認是空的）
+docs （index,replease版本里面沒有文檔）
+example （幾個例子）
+lib （activeMQ使用到的lib）
+webapps （系統管理員控制檯代碼）
+webapps-demo（系統示例代碼）
-activemq-all-5.8.0.jar (ActiveMQ的binary)
-user-guide.html （部署指引）
-LICENSE.txt
-NOTICE.txt
-README.txt
其餘文件就不相信介紹了，搞Java的應該都知道幹什麼用的。
你能夠進入bin目錄，使用activemq.bat雙擊啓動（windows用戶能夠選擇系統位數，若是你是linux的話，就用命令行的發送去啓動），若是一切順利，你就會看見相似下面的信息：
若是你看到這個，那麼恭喜你成功了。若是你啓動看到了異常信息：
Caused by: java.io.IOException: Failed to bind to server socket: tcp://0.0.0.0:61616?maximumConnections=1000&wireformat.maxFrameSize=104857600 due to: java.net.SocketException: Unrecognized Windows Sockets error: 0: JVM_Bind
那麼我告訴你，很不幸，你的端口被佔用了。接下來你大概想知道是哪一個程序佔用了你的端口，並kill掉該進程或服務。或者你要嘗試修改ActiveMQ的默認端口61616（ActiveMQ使用的默認端口是61616），在大多數狀況下，佔用61616端口的是Internet Connection Sharing (ICS) 這個Windows服務，你只需中止它就能夠啓動ActiveMQ了。
四、 啓動成功就能夠訪問管理員界面：http://localhost:8161/admin，默認用戶名和密碼admin/admin。若是你想修改用戶名和密碼的話，在conf/jetty-realm.properties中修改便可。
其中在導航菜單中，Queues是隊列方式消息。Topics是主題方式消息。Subscribers消息訂閱監控查詢。Connections能夠查看連接數，分別能夠查看xmpp、ssl、stomp、openwire、ws和網絡連接。Network是網絡連接數監控。Send能夠發送消息數據。
五、 運行demo示例，在dos控制檯輸入activemq.bat xbean:../conf/activemq-demo.xml 便可啓動demo示例。官方提供的user-guide.html中的access the web console中是提示輸入：activemq.bat console xbean:conf/activemq-demo.xml，我用這種方式不成功。
固然你還能夠用絕對的文件目錄方式：activemq.bat xbean:file:D:/mq/conf/activemq-demo.xml
若是提示conf/activemq-demo.xml沒有找到，你能夠嘗試改變下路徑，也就是去掉上面的「..」。經過http://localhost:8161/demo/ 就能夠訪問示例了。
4、 消息示例
JMS 公共
點對點域
發佈/訂閱域
ConnectionFactory
QueueConnectionFactory
TopicConnectionFactory
Connection
QueueConnection
TopicConnection
Destination
Queue
Topic
Session
QueueSession
TopicSession
MessageProducer
QueueSender
TopicPublisher
MessageConsumer
QueueReceiver
TopicSubscriber
(1)、點對點方式（point-to-point）
點對點的消息發送方式主要創建在 Message Queue,Sender,reciever上，Message Queue 存貯消息，Sneder 發送消息，receive接收消息.具體點就是Sender Client發送Message Queue ,而 receiver Cliernt從Queue中接收消息和"發送消息已接受"到Quere,確認消息接收。消息發送客戶端與接收客戶端沒有時間上的依賴，發送客戶端能夠在任什麼時候刻發送信息到Queue，而不須要知道接收客戶端是否是在運行
(2)、發佈/訂閱 方式（publish/subscriber Messaging）
發佈/訂閱方式用於多接收客戶端的方式.做爲發佈訂閱的方式，可能存在多個接收客戶端，而且接收端客戶端與發送客戶端存在時間上的依賴。一個接收端只能接收他建立之後發送客戶端發送的信息。做爲subscriber ,在接收消息時有兩種方法，destination的receive方法，和實現message listener 接口的onMessage 方法。
二、ActiviteMQ接收和發送消息基本流程
發送消息的基本步驟：
(1)、建立鏈接使用的工廠類JMS ConnectionFactory
(2)、使用管理對象JMS ConnectionFactory創建鏈接Connection，並啓動
(3)、使用鏈接Connection 創建會話Session
(4)、使用會話Session和管理對象Destination建立消息生產者MessageSender
(5)、使用消息生產者MessageSender發送消息
消息接收者從JMS接受消息的步驟
(1)、建立鏈接使用的工廠類JMS ConnectionFactory
(2)、使用管理對象JMS ConnectionFactory創建鏈接Connection，並啓動
(3)、使用鏈接Connection 創建會話Session
(4)、使用會話Session和管理對象Destination建立消息接收者MessageReceiver
(5)、使用消息接收者MessageReceiver接受消息，須要用setMessageListener將MessageListener接口綁定到MessageReceiver消息接收者必須實現了MessageListener接口，須要定義onMessage事件方法。
5、 代碼示例
在代碼開始，咱們先建一個project，在這個project中添加以下jar包
添加完jar包後就能夠開始實際的代碼工做了。
但對於我我的而言，其實不必費勁巴拉的加入這三個包，你能夠直接加入下面紅色圈出來的jar，包含了mq開發的全部jar文件，方便快捷。
ActiviteMQ消息有3中形式，
一、JMS公共
二、點對點域
三、發佈/訂閱域
它們的前半部分基本上差很少，建立鏈接工廠，獲取相應的鏈接，獲取相應的session，主要的區別在於
一、JMS公共 使用MessageProducer發送消息
二、點對點域 使用QueueSender發送消息
三、發佈/訂閱域 使用TopicPublisher發送消息
若是你查看QueueSender和TopicPublisher的源碼，你會發現，它們都是MessageProducer接口的子接口,其它你懂的~
一、 使用JMS方式發送接收消息
消息發送者，消息的發送者
package com.xx.mq.jms; 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; /** * 消息發送者 * @author hao * @since 2015-9-24 * @file MessageSender.java * @package com.xx.mq.jms * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class MessageSender { // 發送次數 public static final int SEND_NUM = 10; // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.jms"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null /** * 發送消息 * @param session * @param producer * @throws Exception */ public static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 0; i < SEND_NUM; i++) { String message = "發送消息第" + (i + 1) + "條"; TextMessage text = session.createTextMessage(message); System.out.println(message); producer.send(text); } //生產者睡眠100s // Thread.sleep(100000); } public static void run() throws Exception { Connection connection = null; Session session = null; try { // 一、建立連接工廠 ConnectionFactory factory = new ActiveMQConnectionFactory(MessageReceiver_Listener.DEFAULT_USER, MessageReceiver_Listener.DEFAULT_PASSWORD, MessageReceiver_Listener.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Destination destination = session.createQueue(DESTINATION); // 六、建立消息生產者 MessageProducer producer = session.createProducer(destination); // 七、設置持久化模式 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); //發送消息 sendMessage(session, producer); // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { MessageSender.run(); } }
消息的接收者——————使用Receive的接收方式，在實際的開發中使用比較少。
package com.xx.mq.jms; 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.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息發送者 * @author hao * @since 2015-9-24 * @file MessageReceiver.java * @package com.xx.mq.jms * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class MessageReceiver_Receive { // 發送次數 public static final int SEND_NUM = 10; // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.jms"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null public static void run() throws Exception { Connection connection = null; Session session = null; try { // 建立連接工廠 ConnectionFactory factory = new ActiveMQConnectionFactory(MessageReceiver_Receive.DEFAULT_USER, MessageReceiver_Receive.DEFAULT_PASSWORD, MessageReceiver_Receive.BROKER_URL); // 經過工廠建立一個鏈接 connection = factory.createConnection(); // 啓動鏈接 connection.start(); // 建立一個session會話 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 建立一個消息隊列 Destination destination = session.createQueue(DESTINATION); // 建立消息製做者 MessageConsumer consumer = session.createConsumer(destination); while (true) { // 接收數據的時間（等待） 10 s ,超過10秒自動超時 Message message = consumer.receive(1000*10); // 接收相關的消息 TextMessage text = (TextMessage) message; if (text != null) { System.out.println("接收：" + text.getText()); } else { break; } } // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { MessageReceiver_Receive.run(); } }
消息的接收者——————使用Listener的接收方式，在實際的開發中使用比較多。package com.xx.mq.jms; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageListener; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息發送者 * @author hao * @since 2015-9-24 * @file MessageReceiver.java * @package com.xx.mq.jms * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class MessageReceiver_Listener { // 發送次數 public static final int SEND_NUM = 10; // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.jms"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null public static void run() throws Exception { Connection connection = null; Session session = null; try { // 一、建立連接工廠 ConnectionFactory factory = new ActiveMQConnectionFactory(MessageReceiver_Listener.DEFAULT_USER, MessageReceiver_Listener.DEFAULT_PASSWORD, MessageReceiver_Listener.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Destination destination = session.createQueue(DESTINATION); // 建立消息製做者 MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message msg) { // 接收相關的消息 TextMessage text = (TextMessage) msg; if (text != null) { try { System.out.println("接收：" + text.getText()); } catch (JMSException e) { e.printStackTrace(); } } } }); //讓當前主線程睡眠10秒，單可是監聽器會繼續執行 Thread.sleep(1000*10); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { MessageReceiver_Listener.run(); } }
運行結果展現:
一、點擊運行MessageSender的主方法，查看服務器控制檯下面的 http://localhost:8161/admin/queues.jsp
能夠看到下面的結果:
二、運行MessageReceiver_Listener或者MessageReceiver_Receive的main方法，便可在控制檯看到下面的結果，表示消費者消費成功。
二、 Queue隊列方式發送點對點消息數據
發送方，對於這種點對點的發送方式而言，一旦服務器中的消息被相應的消費者消費後，對應的消息就會馬上被移出消息隊列。
package com.xx.mq.queue; import javax.jms.DeliveryMode; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueSession; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 點對點消息發送者 * @author hao * @since 2015-9-24 * @file QueueSender.java * @package com.xx.mq.queue * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class QueueSender { // 發送次數 public static final int SEND_NUM = 10; // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.queue"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null /** * 發送消息 * @param session * @param sender * @throws Exception */ public static void sendMessage(QueueSession session, javax.jms.QueueSender sender) throws Exception { for (int i = 0; i < SEND_NUM; i++) { String message = "發送第" + (i + 1) + "條消息"; TextMessage textMessage = session.createTextMessage(message); System.out.println(textMessage.getText()); sender.send(textMessage); } } /** * 建立鏈接併發送消息 * @throws Exception */ public static void run() throws Exception { //點對點隊列鏈接 QueueConnection connection = null; //點對點會話Session QueueSession session = null; try { // 一、建立連接工廠 QueueConnectionFactory factory = new ActiveMQConnectionFactory(QueueSender.DEFAULT_USER, QueueSender.DEFAULT_PASSWORD, QueueSender.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createQueueConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Queue queue = session.createQueue(DESTINATION); // 六、建立消息發送者 javax.jms.QueueSender sender = session.createSender(queue); // 設置持久化模式 sender.setDeliveryMode(DeliveryMode.NON_PERSISTENT); sendMessage(session, sender); // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { QueueSender.run(); } }
接收方——————使用Receive的方式接收相關的消息，在實際開發中使用的比較少。
package com.xx.mq.queue; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueSession; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 點對點消息接收者 直接Receive 方式 * @author hao * @since 2015-9-24 * @file QueueReceiver_Receive.java * @package com.xx.mq.queue * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class QueueReceiver_Receive { // 接收消息的個數 public static final int Receive_NUM = 10; // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.queue"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null public static void run() throws Exception { QueueConnection connection = null; QueueSession session = null; try { // 一、建立連接工廠 QueueConnectionFactory factory = new ActiveMQConnectionFactory(QueueReceiver_Receive.DEFAULT_USER, QueueReceiver_Receive.DEFAULT_PASSWORD,QueueReceiver_Receive.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createQueueConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Queue queue = session.createQueue(DESTINATION); // 建立消息接收者 javax.jms.QueueReceiver receiver = session.createReceiver(queue); // 直接Receive 方式 接收消息 for(int i=0;i<QueueReceiver_Receive.Receive_NUM;i++){ TextMessage textMessage=(TextMessage) receiver.receive(); if(textMessage!=null) System.out.println("接收#" + textMessage.getText()); } // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { // 關閉會話 session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { QueueReceiver_Receive.run(); } }接收方——————使用Listener監聽者的方式接收相關的消息，在實際開發中使用比較多。
package com.xx.mq.queue; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueSession; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 點對點消息接收者 使用Listener 監聽方式 在實際項目開發中使用比較多 * @author hao * @since 2015-9-24 * @file QueueReceiver_Listener.java * @package com.xx.mq.queue * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class QueueReceiver_Listener { // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.queue"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null public static void run() throws Exception { QueueConnection connection = null; QueueSession session = null; try { // 一、建立連接工廠 QueueConnectionFactory factory = new ActiveMQConnectionFactory(QueueReceiver_Listener.DEFAULT_USER, QueueReceiver_Listener.DEFAULT_PASSWORD,QueueReceiver_Listener.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createQueueConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Queue queue = session.createQueue(DESTINATION); // 建立消息接收者 javax.jms.QueueReceiver receiver = session.createReceiver(queue); //使用內部類爲消息接收者加載相應的Listener監聽 receiver.setMessageListener(new MessageListener() { //重寫onMessage方法 public void onMessage(Message msg) { if (msg != null) { TextMessage textMessage = (TextMessage) msg; try { System.out.println("接收#" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } } }); // 休眠10s再關閉 接收生產者發送的所有的10條消息 // 須要注意的是這裏使用sleep會使當前正在執行的線程進入休眠狀態 // 也就是QueueReceiver_Listener這個類進入休眠狀態了，而接收者的監聽器仍然會繼續執行的哦。 Thread.sleep(1000 * 10); // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { QueueReceiver_Listener.run(); } }運行結果展現:
一、點擊運行QueueSender的主方法，查看服務器控制檯下面的 http://localhost:8161/admin/queues.jsp
能夠看到下面的結果:
二、運行QueueReceiver_Listener或者QueueReceiver_Receive的main方法，便可在控制檯看到下面的結果，表示消費者消費成功。
三、 Topic主題發佈和訂閱消息
消息發送方:
package com.xx.mq.topic; import javax.jms.DeliveryMode; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicConnectionFactory; import javax.jms.TopicPublisher; import javax.jms.TopicSession; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 發佈訂閱式消息發送者 * @author hao * @since 2015-9-24 * @file TopicProducer.java * @package com.xx.mq.topic * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class TopicProducer { // 發送次數 public static final int SEND_NUM = 10; // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/topics.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.topic"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null /** * 消息發送端 * @param session * @param publisher * @throws Exception */ public static void sendMessage(TopicSession session, TopicPublisher publisher) throws Exception { for (int i = 0; i < SEND_NUM; i++) { String message = "發送消息第" + (i + 1) + "條"; TextMessage textMessage = session.createTextMessage(message); System.out.println(textMessage.getText()); //發送 Topic消息 publisher.send(textMessage); } } public void run() throws Exception { //Topic鏈接 TopicConnection connection = null; //Topic會話 TopicSession session = null; try { // 一、建立連接工廠 TopicConnectionFactory factory = new ActiveMQConnectionFactory(TopicProducer.DEFAULT_USER, TopicProducer.DEFAULT_PASSWORD, TopicProducer.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createTopicConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Topic topic = session.createTopic(DESTINATION); // 六、建立消息發送者 TopicPublisher publisher = session.createPublisher(topic); // 設置持久化模式 publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT); sendMessage(session, publisher); // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { new TopicProducer().run(); } }
接收方——————使用Receive的方式對消息進行接收，在實際開發中使用的比較少。
package com.xx.mq.topic; import javax.jms.JMSException; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicConnectionFactory; import javax.jms.TopicSession; import javax.jms.TopicSubscriber; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 發佈訂閱式消息接收者 * @author hao * @since 2015-9-24 * @file TopicReceiver_Receive.java * @package com.xx.mq.topic * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class TopicReceiver_Receive { // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/topics.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.topic"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null public static void run() throws Exception { TopicConnection connection = null; TopicSession session = null; try { // 一、建立連接工廠 TopicConnectionFactory factory = new ActiveMQConnectionFactory(TopicReceiver_Receive.DEFAULT_USER, TopicReceiver_Receive.DEFAULT_PASSWORD, TopicReceiver_Receive.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createTopicConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Topic topic = session.createTopic(DESTINATION); // 六、建立消息製做者 final TopicSubscriber subscriber = session.createSubscriber(topic); //接收Topic生產者發送過來的消息 //須要注意的是此處須要啓動一個新的線程來處理問題 new Thread(){ public void run(){ TextMessage textMessage = null; try { while(true){//持續接收消息 textMessage = (TextMessage) subscriber.receive(); if(textMessage==null) break; System.out.println("接收#" + textMessage.getText()); } } catch (JMSException e) { e.printStackTrace(); } } }.start(); // 休眠100s再關閉 接收生產者發送的所有的10條消息 // 須要注意的是這裏使用sleep會使當前正在執行的線程進入休眠狀態 // 也就是TopicReceiver_Receive這個類進入休眠狀態了，而接收者.start方法剛剛啓動的新線程會繼續執行的哦。 Thread.sleep(1000 *100); // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { TopicReceiver_Receive.run(); } }接收方——————使用的方式Listener監聽器的對消息進行接收，在實際開發中使用的比較多。
package com.xx.mq.topic; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicConnectionFactory; import javax.jms.TopicSession; import javax.jms.TopicSubscriber; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 發佈訂閱式消息接收者 * @author hao * @since 2015-9-24 * @file TopicReceiver_Listener.java * @package com.xx.mq.topic * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class TopicReceiver_Listener { // tcp 地址 服務器器端地址 public static final String BROKER_URL =ActiveMQConnection.DEFAULT_BROKER_URL; // 其值爲 "tcp://localhost:61616"; // 目標地址，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/topics.jsp中能夠查詢到發送的mq消息 public static final String DESTINATION = "mq.topic"; //測試鏈接使用默認的用戶名 public static final String DEFAULT_USER = ActiveMQConnection.DEFAULT_USER;//默認爲null //測試鏈接使用默認的密碼 public static final String DEFAULT_PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默認爲null public static void run() throws Exception { TopicConnection connection = null; TopicSession session = null; try { // 一、建立連接工廠 TopicConnectionFactory factory = new ActiveMQConnectionFactory(TopicReceiver_Listener.DEFAULT_USER, TopicReceiver_Listener.DEFAULT_PASSWORD, TopicReceiver_Listener.BROKER_URL); // 二、經過工廠建立一個鏈接 connection = factory.createTopicConnection(); // 三、啓動鏈接 connection.start(); // 四、建立一個session會話 session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 五、建立一個消息隊列 Topic topic = session.createTopic(DESTINATION); // 六、建立消息製做者 TopicSubscriber subscriber = session.createSubscriber(topic); //使用監聽器的方式訂閱消息 subscriber.setMessageListener(new MessageListener() { public void onMessage(Message msg) { if (msg != null) { TextMessage textMessage = (TextMessage) msg; try { System.out.println("接收#" + textMessage.getText()); } catch (Exception e) { e.printStackTrace(); } } } }); // 休眠100s再關閉 接收生產者發送的所有的10條消息 // 須要注意的是這裏使用sleep會使當前正在執行的線程進入休眠狀態 // 也就是TopicReceiver_Listener這個類進入休眠狀態了，而接收者的監聽器仍然會繼續執行的哦。 Thread.sleep(1000 *100); // 提交會話 session.commit(); } catch (Exception e) { throw e; } finally { // 關閉釋放資源 if (session != null) { session.close(); } if (connection != null) { connection.close(); } } } public static void main(String[] args) throws Exception { TopicReceiver_Listener.run(); } }
運行結果展現:
一、首先啓動消費者，須要注意的是發佈訂閱式的消息模式同點對點消費的模式不一樣。
對於點對點的通訊模式而言，咱們須要首先啓動生產者，將消息發送到服務器上的消息隊列上面，最後啓動生產者，進行消息的消費操做。
對於發佈訂閱式的通訊模式而言，消費者必須先於生產者而啓動，否者的話，消費者是接收不到相應的消費消息的。
相應的訪問鏈接爲: http://localhost:8161/admin/topics.jsp
啓動消費者後（這裏咱們分別啓動Receive和Listener兩個消費者，模擬一個主題多個消費者的狀況，代碼中咱們設置消費者hold的時間爲100s，超時後消費者將自動關閉，若是您的反射弧太長
建議設置爲更大值，ps開個玩笑。），咱們能夠在服務器上看到下面的內容：一、首先啓動生產者，一旦生成這啓動以後，咱們將會在控制檯看到消息的消費狀況。
經過這個控制檯，咱們清楚的看到同一個主題被兩個消費者消費了。須要注意的是啓動生產者後，不會馬上看到上面狀況，須要等100s以後兩個消費者都commit以後才能看到上面的狀況。
四、 整合Spring實現消息發送和接收，在整合以前咱們須要添加jar包，須要的jar包以下
下面這個是我我的在測試開發的時候使用的jar文件，僅供參考。
這些jar包能夠在D:\apache-activemq-5.8.0\lib這個lib目錄中找到，添加完jar包後就開始編碼工做。
消息發送者
package com.xx.mq.spring; import java.text.SimpleDateFormat; import java.util.Date; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; import javax.jms.TextMessage; import org.springframework.context.ApplicationContext; import org.springframework.context.support.FileSystemXmlApplicationContext; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; /** * Spring JMSTemplate 消息發送者<br> * 將JMS整合到spring上面進行開發 * @author hao * @since 2015-9-24 * @file MessageReceiver.java * @package com.xx.mq.jms * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class Sender { public static void main(String[] args) { ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:applicationContext.xml"); JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate"); jmsTemplate.send(new MessageCreator() { public Message createMessage(Session session) throws JMSException { TextMessage message = session.createTextMessage(); SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); String date=formatter.format(new Date()); String text="current system time: "+date; message.setText(text); System.out.println(text); return message; } }); } }
消息接收者
package com.xx.mq.spring; import javax.jms.JMSException; import javax.jms.TextMessage; import org.springframework.context.ApplicationContext; import org.springframework.context.support.FileSystemXmlApplicationContext; import org.springframework.jms.core.JmsTemplate; /** * Spring JMSTemplate 消息接收者<br> * 將JMS整合到spring上面進行開發 * @author hao * @since 2015-9-24 * @file MessageReceiver.java * @package com.xx.mq.jms * @project ActiveMQ-5.8 * @blog http://cnblogs.com/xiohao * @email 1261245595@qq.com * @version 1.0 */ public class Receiver { public static void main(String[] args) throws JMSException { ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:applicationContext.xml"); JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate"); while(true) { //設置10s超時時間 jmsTemplate.setReceiveTimeout(1000*10); TextMessage text = (TextMessage) jmsTemplate.receive(); if(text==null) break; //接收到相應的消息 System.out.println("收到消息：" + text.getText()); } } }
這裏主要是用到了JmsTemplate這個消息模板，這個對象在spring的IoC容器中管理，因此要從spring的容器上下文中獲取。下面看看spring的配置文件applicationContext-beans.xml內容：
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd"> <!-- 鏈接池 <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory"> <bean class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616" /> </bean> </property> </bean> --> <!-- 鏈接工廠 --> <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616" /> </bean> <!-- 配置消息目標 --> <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue"> <!-- 目標，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp --> <constructor-arg index="0" value="mq.queue" /> </bean> <!-- 消息模板 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="activeMQConnectionFactory" /> <property name="defaultDestination" ref="destination" /> <property name="messageConverter"> <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" /> </property> </bean> </beans>運行效果同以前的效果是一一致的，這裏就不進行展現了~~~
這裏的整合就比較簡單了，若是你是web工程，那你在須要用jms的時候，只需用注入jmsTemplate便可。
到此mq的基礎入門講解已經所有講解完畢了。有任何問題，歡迎留言探討~~
參考連接：http://www.cnblogs.com/linjiqin/category/486898.html
http://blog.csdn.net/zbw18297786698/article/category/6496088
相關文章
- 1. 深刻淺出 消息隊列 ActiveMQ
- 2. 深入淺出 消息隊列 ActiveMQ (轉)
- 3. 消息隊列activemq
- 4. 深刻淺出JMS(六)--ActiveMQ實現消息隊列發送郵件
- 5. ActiveMQ消息隊列
- 6. 消息隊列-ActiveMQ
- 7. 消息隊列ActiveMQ
- 8. 消息隊列—ActiveMQ
- 9. ActiveMq消息隊列
- 10. 消息隊列 activeMQ
- 更多相關文章...
- • C# 隊列（Queue） - C#教程
- • HTTP 消息結構 - HTTP 教程
- • 算法總結-深度優先算法
- • 使用阿里雲OSS+CDN部署前端頁面與加速靜態資源
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 深刻淺出 消息隊列 ActiveMQ
- 2. 深入淺出 消息隊列 ActiveMQ (轉)
- 3. 消息隊列activemq
- 4. 深刻淺出JMS(六)--ActiveMQ實現消息隊列發送郵件
- 5. ActiveMQ消息隊列
- 6. 消息隊列-ActiveMQ
- 7. 消息隊列ActiveMQ
- 8. 消息隊列—ActiveMQ
- 9. ActiveMq消息隊列
- 10. 消息隊列 activeMQ