深入淺出消息隊列 ActiveMQ------增強版

本文轉載自查看原文 2015-09-24 15:56 6147 Java SE（Java基礎篇）/ ActiveMQ+JMQ+IBM MQ+KAFKA/ Java EE（Java企業篇）

　　本小節我們將講解Apache開源下的ActiveMQ，而ActiveMQ是JMS的一個具體實現。JMS即Java消息服務（Java Message Service）應用程序接口是一個Java平台中關於面向消息中間件（MOM）的API，用於在兩個應用程序之間，或分布式系統中發送消息，進行異步通信。Java消息服務是一個與具體平台無關的API，絕大多數MOM提供商都對JMS提供支持。

在學習ActiveMQ之前，我們有必要對JMS進行一下簡單的了解，由於篇幅的限制，對於JMS的學習請參考下面的鏈接:

JMS參考鏈接點擊這里

一、概述與介紹

ActiveMQ 是Apache出品，最流行的、功能強大的即時通訊和集成模式的開源服務器。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現。提供客戶端支持跨語言和協議，帶有易於在充分支持JMS 1.1和1.4使用J2EE企業集成模式和許多先進的功能。

二、特性

1、多種語言和協議編寫客戶端。語言： Java、C、C++、C#、Ruby、Perl、Python、PHP。應用協議：OpenWire、Stomp REST、WS Notification、XMPP、AMQP

2、完全支持JMS1.1和J2EE 1.4規范（持久化，XA消息，事務)

3、對Spring的支持，ActiveMQ可以很容易內嵌到使用Spring的系統里面去，而且也支持Spring2.0的特性

4、通過了常見J2EE服務器（如 Geronimo、JBoss 4、GlassFish、WebLogic)的測試，其中通過JCA 1.5 resource adaptors的配置，可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上

5、支持多種傳送協議：in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA

6、支持通過JDBC和journal提供高速的消息持久化

7、從設計上保證了高性能的集群，客戶端-服務器，點對點

8、支持Ajax

9、支持與Axis的整合

10、可以很容易得調用內嵌JMS provider，進行測試

對於Active MQ更多的信息可以點擊這里

三、安裝

開發環境：

System：Windows

JDK：1.6+

IDE：myeclipse

apache ActiveMQ 5.8

Email：1261245595@qq.com

Blog：http://www.cnblogs.com/xiohao

1、下載ActiveMQ，下載地址：http://www.apache.org/dyn/closer.cgi?path=/activemq/apache-activemq/5.8.0/apache-activemq-5.8.0-bin.zip

2、解壓apache-activemq-5.8.0.zip即可完成ActiveMQ的安裝

3、解壓后目錄結構如下

+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了。

4、啟動成功就可以訪問管理員界面：http://localhost:8161/admin，默認用戶名和密碼admin/admin。如果你想修改用戶名和密碼的話，在conf/jetty-realm.properties中修改即可。

其中在導航菜單中，Queues是隊列方式消息。Topics是主題方式消息。Subscribers消息訂閱監控查詢。Connections可以查看鏈接數，分別可以查看xmpp、ssl、stomp、openwire、ws和網絡鏈接。Network是網絡鏈接數監控。Send可以發送消息數據。

5、運行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/ 就可以訪問示例了。

四、消息示例

1、ActiviteMQ消息有3中形式

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 方法。

2、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事件方法。

五、代碼示例

在代碼開始，我們先建一個project，在這個project中添加如下jar包

添加完jar包后就可以開始實際的代碼工作了。

但對於我個人而言，其實沒必要費勁巴拉的加入這三個包，你可以直接加入下面紅色圈出來的jar，包含了mq開發的所有jar文件，方便快捷。

ActiviteMQ消息有3中形式，

1、JMS公共

2、點對點域

3、發布/訂閱域

它們的前半部分基本上差不多，創建連接工廠，獲取相應的連接，獲取相應的session，主要的區別在於

1、JMS公共使用MessageProducer發送消息

2、點對點域使用QueueSender發送消息

3、發布/訂閱域使用TopicPublisher發送消息

如果你查看QueueSender和TopicPublisher的源碼，你會發現，它們都是MessageProducer接口的子接口,其它你懂的~

1、使用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 {
	            // 1、創建鏈接工廠
	            ConnectionFactory factory = new ActiveMQConnectionFactory(MessageReceiver_Listener.DEFAULT_USER, MessageReceiver_Listener.DEFAULT_PASSWORD, MessageReceiver_Listener.BROKER_URL);
	            // 2、通過工廠創建一個連接
	            connection = factory.createConnection();
	            // 3、啟動連接 
	            connection.start();
	            // 4、創建一個session會話
	            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
	            // 5、創建一個消息隊列
	            Destination destination = session.createQueue(DESTINATION);
	            // 6、創建消息生產者
	            MessageProducer producer = session.createProducer(destination);
	            // 7、設置持久化模式
	            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 {
            // 1、創建鏈接工廠
            ConnectionFactory factory = new ActiveMQConnectionFactory(MessageReceiver_Listener.DEFAULT_USER, MessageReceiver_Listener.DEFAULT_PASSWORD, MessageReceiver_Listener.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            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();
    }
}

運行結果展示:

1、點擊運行MessageSender的主方法，查看服務器控制台下面的 http://localhost:8161/admin/queues.jsp

可以看到下面的結果:

2、運行MessageReceiver_Listener或者MessageReceiver_Receive的main方法，即可在控制台看到下面的結果，表示消費者消費成功。

2、 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 {
            // 1、創建鏈接工廠
            QueueConnectionFactory factory = new ActiveMQConnectionFactory(QueueSender.DEFAULT_USER, QueueSender.DEFAULT_PASSWORD, QueueSender.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createQueueConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            Queue queue = session.createQueue(DESTINATION);
            // 6、創建消息發送者
            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 {
            // 1、創建鏈接工廠
            QueueConnectionFactory factory = new ActiveMQConnectionFactory(QueueReceiver_Receive.DEFAULT_USER, QueueReceiver_Receive.DEFAULT_PASSWORD,QueueReceiver_Receive.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createQueueConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            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 {
            // 1、創建鏈接工廠
            QueueConnectionFactory factory = new ActiveMQConnectionFactory(QueueReceiver_Listener.DEFAULT_USER, QueueReceiver_Listener.DEFAULT_PASSWORD,QueueReceiver_Listener.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createQueueConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            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();
    }
}

運行結果展示:

1、點擊運行QueueSender的主方法，查看服務器控制台下面的 http://localhost:8161/admin/queues.jsp

可以看到下面的結果:

2、運行QueueReceiver_Listener或者QueueReceiver_Receive的main方法，即可在控制台看到下面的結果，表示消費者消費成功。

3、 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 {
            // 1、創建鏈接工廠
            TopicConnectionFactory factory = new ActiveMQConnectionFactory(TopicProducer.DEFAULT_USER, TopicProducer.DEFAULT_PASSWORD, TopicProducer.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createTopicConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            Topic topic = session.createTopic(DESTINATION);
            // 6、創建消息發送者
            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 {
            // 1、創建鏈接工廠
            TopicConnectionFactory factory = new ActiveMQConnectionFactory(TopicReceiver_Receive.DEFAULT_USER, TopicReceiver_Receive.DEFAULT_PASSWORD, TopicReceiver_Receive.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createTopicConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            Topic topic = session.createTopic(DESTINATION);
            // 6、創建消息制作者
            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 {
            // 1、創建鏈接工廠
            TopicConnectionFactory factory = new ActiveMQConnectionFactory(TopicReceiver_Listener.DEFAULT_USER, TopicReceiver_Listener.DEFAULT_PASSWORD, TopicReceiver_Listener.BROKER_URL);
            // 2、通過工廠創建一個連接
            connection = factory.createTopicConnection();
            // 3、啟動連接
            connection.start();
            // 4、創建一個session會話
            session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 5、創建一個消息隊列
            Topic topic = session.createTopic(DESTINATION);
            // 6、創建消息制作者
            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();
    }
}

運行結果展示:
1、首先啟動消費者，需要注意的是發布訂閱式的消息模式同點對點消費的模式不同。
對於點對點的通信模式而言，我們需要首先啟動生產者，將消息發送到服務器上的消息隊列上面，最后啟動生產者，進行消息的消費操作。
對於發布訂閱式的通信模式而言，消費者必須先於生產者而啟動，否者的話，消費者是接收不到相應的消費消息的。
相應的訪問連接為:　　http://localhost:8161/admin/topics.jsp
啟動消費者后（這里我們分別啟動Receive和Listener兩個消費者，模擬一個主題多個消費者的情況，代碼中我們設置消費者hold的時間為100s，超時后消費者將自動關閉，如果您的反射弧太長
建議設置為更大值，ps開個玩笑。），我們可以在服務器上看到下面的內容：

1、首先啟動生產者，一旦生成這啟動之后，我們將會在控制台看到消息的消費情況。


通過這個控制台，我們清楚的看到同一個主題被兩個消費者消費了。需要注意的是啟動生產者后，不會立刻看到上面情況，需要等100s之后兩個消費者都commit之后才能看到上面的情況。

4、整合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

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 深入淺出消息隊列 ActiveMQ (轉) 深入淺出 JMS(四) - ActiveMQ 消息存儲深入淺出AQS之條件隊列 [Cocoa]深入淺出Cocoa之消息深入淺出 HTTPS (詳解版) 深入淺出之切空間深入淺出電位 ArrayList 深入淺出深入淺出KnockoutJS Docker的深入淺出

JMS 公共	點對點域	發布/訂閱域
ConnectionFactory	QueueConnectionFactory	TopicConnectionFactory
Connection	QueueConnection	TopicConnection
Destination	Queue	Topic
Session	QueueSession	TopicSession
MessageProducer	QueueSender	TopicPublisher
MessageConsumer	QueueReceiver	TopicSubscriber

深入淺出 消息隊列 ActiveMQ------增強版

免責聲明！

深入淺出消息隊列 ActiveMQ------增強版