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

?

前言

軟件系統(tǒng)的發(fā)展，簡(jiǎn)單來(lái)說(shuō)是從單機(jī)應(yīng)用逐步演變?yōu)? 分布式+集群 模式，當(dāng)然這根互聯(lián)網(wǎng)的興起是密不可分的。筆者曾經(jīng)面試過(guò)一些開發(fā)人員，當(dāng)你們使用代碼編寫一個(gè)簡(jiǎn)單的Login模塊時(shí)，大概需要花費(fèi)多少周期？有些開發(fā)人員不假思索的脫口而出，5分鐘、10分鐘等。OK，我們先不管他們是否真的能夠在這么短的時(shí)間內(nèi)開發(fā)出一個(gè)簡(jiǎn)單的Login功能，我真正想問(wèn)的是，這個(gè)看似簡(jiǎn)單的功能背后，如果存放在互聯(lián)網(wǎng)場(chǎng)景下，如何能夠保證系統(tǒng)同時(shí)承載100——100W人的 并發(fā)洗禮 而不會(huì)出現(xiàn)宕機(jī)？緊接著是死一般的 沉寂 。

?

目錄

一、分布式系統(tǒng)與消息中間件

二、JMS與ActiveMQ

三、ActiveMQ的下載與啟動(dòng)

四、消息的傳遞模型

五、Queue與Topic

六、補(bǔ)充——消息流程

?

一、分布式系統(tǒng)與消息中間件

前言這個(gè)故事，或許并沒有筆者說(shuō)得這么夸張，但仔細(xì)想想看，如果業(yè)務(wù)場(chǎng)景對(duì)并發(fā)性、實(shí)時(shí)性要求較高時(shí)，任何一個(gè)看似簡(jiǎn)單的系統(tǒng)功能，都將會(huì)變得 異常復(fù)雜 ，因?yàn)檫@里的復(fù)雜是來(lái)源于多方面的，最主要的就是如果系統(tǒng)因?yàn)槟骋粋€(gè)功能出現(xiàn)異常將會(huì)導(dǎo)致 整個(gè)系統(tǒng)服務(wù)不可用 。那么應(yīng)該怎么辦呢？如果你不想過(guò)多的，或者是在短時(shí)間內(nèi)追加過(guò)多的 投資 ，集群+物理堆機(jī)+適當(dāng)?shù)拇a性能調(diào)優(yōu)，或許是解決問(wèn)題的最好辦法，但這樣真的一勞永逸嗎？我們都知道集群的 威力 從理論上來(lái)說(shuō)會(huì)是 單節(jié)點(diǎn)部署的數(shù)倍 ，但集群如果沒有細(xì)化，也就是說(shuō)，一個(gè)系統(tǒng)的所有業(yè)務(wù)模塊仍然耦合在一起時(shí)，無(wú)論你的集群的規(guī)模有多么龐大，從外部看起來(lái)，仍然只是一個(gè)大雜燴， 凌亂不堪 。

?

圖a 單節(jié)點(diǎn)部署與集群部署

?

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

?

MQ(Message Queue，消息隊(duì)列)， 一切以消息為主 。假設(shè)你現(xiàn)在已經(jīng)將原本冗余的一個(gè)系統(tǒng)根據(jù)業(yè)務(wù)邏輯進(jìn)行了適當(dāng)?shù)牟鸱郑⑶颐恳粋€(gè)子系統(tǒng)都是一個(gè)集群，在這種情況下，由UM子系統(tǒng)處理跟用戶相關(guān)的所有業(yè)務(wù)，但當(dāng)它處理完自身業(yè)務(wù)后，還需要調(diào)用其它子系統(tǒng)的服務(wù)時(shí)應(yīng)該怎么辦？或許某些開發(fā)人員會(huì)覺得這似乎很簡(jiǎn)單，直接調(diào)用啊，一個(gè)長(zhǎng)/短鏈接請(qǐng)求就可以解決。但是這樣一來(lái)，UM子系統(tǒng)將會(huì) 被動(dòng)的依賴 N多的子系統(tǒng)，如果后期再進(jìn)行 追加 ，UM的依賴會(huì)越來(lái)越多，并且這種 同步調(diào)用 的方式在某些情況下似乎并 不高效 。換句話多，UM子系統(tǒng)的職責(zé)僅僅只是負(fù)責(zé)與用戶相關(guān)的所有業(yè)務(wù)，其它業(yè)務(wù)與它無(wú)關(guān)，哪怕是服務(wù)調(diào)用這個(gè)過(guò)程也 不該 由它來(lái)做，如果你也是這么想的，那么恭喜你，你已經(jīng)考慮到 跨系統(tǒng)之間的依賴解耦 和 通過(guò)異步調(diào)用的方式提升程序的響應(yīng)性能 的高度。

?圖b 直接服務(wù)調(diào)用與MQ消息推送

?

如圖b所示，UM子系統(tǒng)的任務(wù)非常明確，當(dāng)處理完用戶登錄任務(wù)后，直接將登錄結(jié)果發(fā)送給MQ服務(wù)器，如果訂閱了該主題的子系統(tǒng)自然會(huì)收到消息，接下來(lái)，各個(gè)子系統(tǒng)該干嘛干嘛去，UM子系統(tǒng)徹底 解耦 。這里有一點(diǎn)筆者需要提出來(lái)，本章的主角ActiveMQ或許并 不太適用于互聯(lián)網(wǎng)場(chǎng)景 ，至于為什么，筆者稍后會(huì)進(jìn)行詳解。

?

二、JMS與ActiveMQ

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

?

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

?

三、ActiveMQ的下載與啟動(dòng)

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

?

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

?

四、消息傳遞模型

之前筆者曾經(jīng)提及過(guò)，ActiveMQ其實(shí)就是JMS規(guī)范的一種實(shí)現(xiàn)，那么按照J(rèn)MS的架構(gòu)模型來(lái)看，JMS Provider的主要任務(wù)就是負(fù)責(zé) 消息路由 ，推送消息的客戶端稱之為生產(chǎn)者(Producer)，而接受消息的客戶端則稱之為消費(fèi)者(Consumer)。JMS的消息傳遞模型有2種，一種是基于點(diǎn)對(duì)點(diǎn)(P oint-to-Point，p2p )的，而另外一種則是基于 廣播形式 的 發(fā)布/訂閱( publish/subscribe， pub/sub ) 。

?

點(diǎn)對(duì)點(diǎn)的消息傳遞比較簡(jiǎn)單，無(wú)非就是一種一對(duì)一的消息推送/消費(fèi)模式，也就是說(shuō)，如果多個(gè)客戶端都在監(jiān)聽消息隊(duì)列上的消息，JMS Provider則會(huì)根據(jù) 先到先得 原則確定唯一的一個(gè)消費(fèi)者，如果沒有客戶端在監(jiān)聽消息隊(duì)列，那么消息將會(huì)保存在消息隊(duì)列中，直到最終確定消費(fèi)者為止，點(diǎn)對(duì)點(diǎn)模型其實(shí)就是一個(gè)典型的 拉模型 (而非Provider主動(dòng)推送)。 發(fā)布/訂閱可以理解為一對(duì)多的消息推送模式，在 發(fā)布/訂閱模型中，只要訂閱了目標(biāo)主題的所有客戶端都可以消費(fèi)生產(chǎn)者推送的消息， pub/sub就是典型的 推模型 ，由JMS Provider主動(dòng)推送消息 。

圖c JMS的消息傳遞模型

?

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

?

五、Queue與Topic

說(shuō)了這么多，是該用代碼進(jìn)行驗(yàn)證的時(shí)候了，那么接下來(lái)筆者將演示如何使用ActiveMQ實(shí)現(xiàn)p2p和 pub/sub等2種消息傳遞模型。在JMS規(guī)范中，Queue實(shí)現(xiàn)了基于P2P的消息傳遞模型，而Topic則實(shí)現(xiàn)了基于 pub/sub的消息傳遞模型。

?

使用Queue實(shí)現(xiàn)基于P2P的消息傳遞模型：

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

	Producer() {
		try {
			/* 創(chuàng)建鏈接工廠 */
			connFactory = new ActiveMQConnectionFactory("admin", "admin",
					"tcp://127.0.0.1:61616");
			/* 獲取鏈接 */
			conn = connFactory.createConnection();
			conn.start();
			/* 獲取Session */
			session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
			/* 創(chuàng)建生產(chǎn)者 */
			producer = session.createProducer(session.createQueue("testQueue"));
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			System.out.println(sendMessage() ? "消息發(fā)送成功..." : "消息發(fā)送成功...");
		} 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 {
				/* 發(fā)送消息 */
				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消費(fèi)者
 *
 * @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實(shí)現(xiàn)了基于P2P的消息傳遞模型，而如果希望使用Topic實(shí)現(xiàn) pub/sub模型，只需要將生產(chǎn)者/消費(fèi)者的代碼片段：

      session.createQueue("testQueue")
    

?

?更改為如下即可：

      session.createTopic("testTopic")
    

?

六、補(bǔ)充——消息流程

?發(fā)送消息的基本步驟：
1、創(chuàng)建連接工廠類；
2、通過(guò)ConnectionFactory建立連接Connection，并啟動(dòng)；
3、使用連接Connection建立會(huì)話Session；
4、使用會(huì)話Session和管理對(duì)象Destination創(chuàng)建消息生產(chǎn)者M(jìn)essageSender；
5、生產(chǎn)者發(fā)送消息；

?

消息接收者從JMS接受消息的步驟
1、創(chuàng)建連接工廠類；
2、通過(guò)ConnectionFactory建立連接Connection，并啟動(dòng)；
3、使用連接Connection建立會(huì)話Session；
4、使用會(huì)話Session和管理對(duì)象Destination創(chuàng)建消息接收者M(jìn)essageReceiver；
5、消費(fèi)者消費(fèi)消息；

?

本章內(nèi)容到此結(jié)束，由于時(shí)間倉(cāng)庫(kù)，本文或許有很多不盡人意的地方，希望各位能夠理解和體諒。另外 http://boy00fly.iteye.com/blog/1103586 也是一篇不錯(cuò)的介紹ActiveMQ的博文。