《筆者帶你剖析輕量級消息中間件ActiveMQ》

?

前言

軟件系統的發展，簡單來說是從單機應用逐步演變為 分布式+集群 模式，當然這根互聯網的興起是密不可分的。筆者曾經面試過一些開發人員，當你們使用代碼編寫一個簡單的Login模塊時，大概需要花費多少周期？有些開發人員不假思索的脫口而出，5分鐘、10分鐘等。OK，我們先不管他們是否真的能夠在這么短的時間內開發出一個簡單的Login功能，我真正想問的是，這個看似簡單的功能背后，如果存放在互聯網場景下，如何能夠保證系統同時承載100——100W人的 并發洗禮 而不會出現宕機？緊接著是死一般的 沉寂 。

?

目錄

一、分布式系統與消息中間件

二、JMS與ActiveMQ

三、ActiveMQ的下載與啟動

四、消息的傳遞模型

五、Queue與Topic

六、補充——消息流程

?

一、分布式系統與消息中間件

前言這個故事，或許并沒有筆者說得這么夸張，但仔細想想看，如果業務場景對并發性、實時性要求較高時，任何一個看似簡單的系統功能，都將會變得 異常復雜 ，因為這里的復雜是來源于多方面的，最主要的就是如果系統因為某一個功能出現異常將會導致 整個系統服務不可用 。那么應該怎么辦呢？如果你不想過多的，或者是在短時間內追加過多的 投資 ，集群+物理堆機+適當的代碼性能調優，或許是解決問題的最好辦法，但這樣真的一勞永逸嗎？我們都知道集群的 威力 從理論上來說會是 單節點部署的數倍 ，但集群如果沒有細化，也就是說，一個系統的所有業務模塊仍然耦合在一起時，無論你的集群的規模有多么龐大，從外部看起來，仍然只是一個大雜燴， 凌亂不堪 。

?

圖a 單節點部署與集群部署

?

那么如果是在 條件允許 的情況下，真正該做的事情是根據不同的業務邏輯，將原本冗余在一起的業務模塊打散、 拆分 成多個子系統，然后進行 模塊化 部署，每一個業務子系統都可以是一個集群，這樣一來，我們的系統才會變得更加 健壯和高效 。當然，分布式環境有分布式環境的好處，但是，同樣也有它的弊端，畢竟系統 依賴 多了，無論是部署還是調試或者是排查問題的時候都會變得無比的復雜和麻煩，本章筆者并不打算教會大家如何根據具體的業務場景構建一個健壯的大型分布式平臺，因此請回到本章的主題——MQ身上。

?

MQ(Message Queue，消息隊列)， 一切以消息為主 。假設你現在已經將原本冗余的一個系統根據業務邏輯進行了適當的拆分，并且每一個子系統都是一個集群，在這種情況下，由UM子系統處理跟用戶相關的所有業務，但當它處理完自身業務后，還需要調用其它子系統的服務時應該怎么辦？或許某些開發人員會覺得這似乎很簡單，直接調用啊，一個長/短鏈接請求就可以解決。但是這樣一來，UM子系統將會 被動的依賴 N多的子系統，如果后期再進行 追加 ，UM的依賴會越來越多，并且這種 同步調用 的方式在某些情況下似乎并 不高效 。換句話多，UM子系統的職責僅僅只是負責與用戶相關的所有業務，其它業務與它無關，哪怕是服務調用這個過程也 不該 由它來做，如果你也是這么想的，那么恭喜你，你已經考慮到 跨系統之間的依賴解耦 和 通過異步調用的方式提升程序的響應性能 的高度。

?圖b 直接服務調用與MQ消息推送

?

如圖b所示，UM子系統的任務非常明確，當處理完用戶登錄任務后，直接將登錄結果發送給MQ服務器，如果訂閱了該主題的子系統自然會收到消息，接下來，各個子系統該干嘛干嘛去，UM子系統徹底 解耦 。這里有一點筆者需要提出來，本章的主角ActiveMQ或許并 不太適用于互聯網場景 ，至于為什么，筆者稍后會進行詳解。

?

二、JMS與ActiveMQ

長話短說，JMS(Java Message Service，Java消息服務)屬于JAVA EE規范之一，同JDBC規范一樣，JMS也僅僅只提供有通用性的接口，而具體的實現，則需要依賴由其遵循了JMS規范的消息中間件負責。ActiveMQ是Apache旗下的一款完全支持JMS1.1和JAVA EE?1.4規范的消息中間件產品，換句話說，ActiveMQ就是JMS規范的實現者之一。

?

之前筆者曾經提及過，ActiveMQ或許并不太適用于互聯網場景，因為JMS規范注定了ActiveMQ更適合于 企業級項目 。在大型的分布式互聯網場景下，跨系統之間的解耦和異步調用這2點，僅僅只是屬于任何一款消息中間件產品 最基礎的特征 ，而大部分情況下，開發人員卻往往更加關注消息的順序、擴展性、可靠性、消息發送的一致性，以及多集群之間訂閱等方面的問題。因此，在大型的分布式互聯網場景下，開發人員更多會選擇MetaQ、Notify、kafka等 天生為互聯網場景而生 的消息中間件產品。或許你會覺得奇怪，為什么筆者會拿ActiveMQ進行開刀？其實最主要的原因是因為其 簡單、輕量、易部署 的特性非常適合作為筆者消息中間件系列博文的 開篇 來進行闡述。

?

三、ActiveMQ的下載與啟動

所謂工欲善其事必先利其器，ActiveMQ的下載地址為 http://activemq.apache.org/download.html ，本章筆者所使用的版本為apache-activemq-5.11.1(for Windows)，當然筆者建議大家下載跟本文一致的版本，這樣既可避免一些不必要的錯誤發生。

?

當成功下載好ActiveMQ后，接下來要做的事情有2個，第一是在目錄“\apache-activemq-5.11.1\bin”中啟動我們的消息服務器( JMS Provider )， 啟動成功就可以訪問管理員界面： http://localhost:8161/admin ，默認用戶名和密碼admin/admin。如果你想修改用戶名和密碼的話，在conf/jetty-realm.properties中修改即可。另外一個便是將目錄“apache-activemq-5.11.1\lib”中的所有構件拷貝到項目工程中，以便于實現MQ客戶端。

?

四、消息傳遞模型

之前筆者曾經提及過，ActiveMQ其實就是JMS規范的一種實現，那么按照JMS的架構模型來看，JMS Provider的主要任務就是負責 消息路由 ，推送消息的客戶端稱之為生產者(Producer)，而接受消息的客戶端則稱之為消費者(Consumer)。JMS的消息傳遞模型有2種，一種是基于點對點(P oint-to-Point，p2p )的，而另外一種則是基于 廣播形式 的 發布/訂閱( publish/subscribe， pub/sub ) 。

?

點對點的消息傳遞比較簡單，無非就是一種一對一的消息推送/消費模式，也就是說，如果多個客戶端都在監聽消息隊列上的消息，JMS Provider則會根據 先到先得 原則確定唯一的一個消費者，如果沒有客戶端在監聽消息隊列，那么消息將會保存在消息隊列中，直到最終確定消費者為止，點對點模型其實就是一個典型的 拉模型 (而非Provider主動推送)。 發布/訂閱可以理解為一對多的消息推送模式，在 發布/訂閱模型中，只要訂閱了目標主題的所有客戶端都可以消費生產者推送的消息， pub/sub就是典型的 推模型 ，由JMS Provider主動推送消息 。

圖c JMS的消息傳遞模型

?

這里筆者還要補充一個內容，JMS的消息體主要由3部分構成，分別是：頭(head)、屬性(property)，以及主體(body)。主體就是具體的 消息內容 ，當然對應不同的消息類型，JMS提供有不同的消息接口，這些消息接口都屬于javax.jms.Message接口，比如： StreamMessage、 MapMessage、 TextMessage、 ObjectMessage、 BytesMessage等。

?

五、Queue與Topic

說了這么多，是該用代碼進行驗證的時候了，那么接下來筆者將演示如何使用ActiveMQ實現p2p和 pub/sub等2種消息傳遞模型。在JMS規范中，Queue實現了基于P2P的消息傳遞模型，而Topic則實現了基于 pub/sub的消息傳遞模型。

?

使用Queue實現基于P2P的消息傳遞模型：

      /**
 * p2p模型生產者
 * 
 * @author gaoxianglong
 */
public class Producer {
	private ConnectionFactory connFactory;
	private Connection conn;
	private Session session;
	private MessageProducer producer;

	Producer() {
		try {
			/* 創建鏈接工廠 */
			connFactory = new ActiveMQConnectionFactory("admin", "admin",
					"tcp://127.0.0.1:61616");
			/* 獲取鏈接 */
			conn = connFactory.createConnection();
			conn.start();
			/* 獲取Session */
			session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
			/* 創建生產者 */
			producer = session.createProducer(session.createQueue("testQueue"));
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			System.out.println(sendMessage() ? "消息發送成功..." : "消息發送成功...");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (null != conn) {
				try {
					conn.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public boolean sendMessage() {
		boolean result = false;
		if (null != session && null != producer) {
			try {
				/* 發送消息 */
				producer.send(session.createTextMessage("test..."));
				session.commit();
				result = true;
			} catch (JMSException e) {
				try {
					session.rollback();
				} catch (JMSException e1) {
					e1.printStackTrace();
				}
			}
		}
		return result;
	}

	public static void main(String[] args) {
		new Producer();
	}
}
    

?

      /**
 * Queue消費者
 *
 * @author gaoxianglong
 */
public class Consumer implements MessageListener {
	private ConnectionFactory connFactory;
	private Connection conn;
	private Session session;
	private MessageConsumer consumer;

	Consumer() {
		try {
			connFactory = new ActiveMQConnectionFactory("admin", "admin",
					"tcp://127.0.0.1:61616");
			conn = connFactory.createConnection();
			conn.start();
			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
			consumer = session.createConsumer(session.createQueue("testQueue"));
			consumer.setMessageListener(this);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		new Consumer();
	}

	@Override
	public void onMessage(Message arg0) {
		TextMessage msg = (TextMessage) arg0;
		try {
			System.out.println("收到消息->" + msg.getText());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
}
    

?

上述程序示例了，筆者使用Queue實現了基于P2P的消息傳遞模型，而如果希望使用Topic實現 pub/sub模型，只需要將生產者/消費者的代碼片段：

      session.createQueue("testQueue")
    

?

?更改為如下即可：

      session.createTopic("testTopic")
    

?

六、補充——消息流程

?發送消息的基本步驟：
1、創建連接工廠類；
2、通過ConnectionFactory建立連接Connection，并啟動；
3、使用連接Connection建立會話Session；
4、使用會話Session和管理對象Destination創建消息生產者MessageSender；
5、生產者發送消息；

?

消息接收者從JMS接受消息的步驟
1、創建連接工廠類；
2、通過ConnectionFactory建立連接Connection，并啟動；
3、使用連接Connection建立會話Session；
4、使用會話Session和管理對象Destination創建消息接收者MessageReceiver；
5、消費者消費消息；

?

本章內容到此結束，由于時間倉庫，本文或許有很多不盡人意的地方，希望各位能夠理解和體諒。另外 http://boy00fly.iteye.com/blog/1103586 也是一篇不錯的介紹ActiveMQ的博文。