高并發(fā)性能的軟件開發(fā)架構(gòu)設(shè)計(jì)咨詢

1、基于消息的分布式架構(gòu)在企業(yè)應(yīng)用系統(tǒng)領(lǐng)域，我們總是會(huì)面對(duì)不同系統(tǒng)之間的通信、集成與整合，尤其當(dāng)面臨異構(gòu)系統(tǒng)時(shí)，這種分布式的調(diào)用與通信變得越重 要，它在架構(gòu)設(shè)計(jì)中就更加凸顯其價(jià)值。并且，從業(yè)務(wù)分析與架構(gòu)質(zhì)量的角度來講，我們也希望在系統(tǒng)架構(gòu)中盡可能地形成對(duì)服務(wù)的重用，通過獨(dú)立運(yùn)行在進(jìn)程中服 務(wù)的形式，徹底解除客戶端與服務(wù)端的耦合。這常常是架構(gòu)演化的必然道路。隨著網(wǎng)絡(luò)基礎(chǔ)設(shè)施的逐步成熟，從RPC進(jìn)化到Web Service，并在業(yè)界開始普遍推行SOA，再到后來的RESTful平臺(tái)以及云計(jì)算中的PaaS與SaaS概念的推廣，分布式架構(gòu)在企業(yè)應(yīng)用中開始呈 現(xiàn)出不同的風(fēng)貌，然而殊途同歸，這些分布式架構(gòu)的目標(biāo)

2、仍然是希望回到建造巴別塔的時(shí)代，系統(tǒng)之間的交流不再為不同語言與平臺(tái)的隔閡而產(chǎn)生障礙。正如 Martin Fowler在企業(yè)集成模式一書的序中寫道：“集成之所以重要是因?yàn)橄嗷オ?dú)立的應(yīng)用是沒有生命力的。我們需要一種技術(shù)能將在設(shè)計(jì)時(shí)并未考慮互操作的應(yīng) 用集成起來，打破它們之間的隔閡，獲得比單個(gè)應(yīng)用更多的效益”。這或許是分布式架構(gòu)存在的主要意義。1、集成模式中的消息模式歸根結(jié)底，企業(yè)應(yīng)用系統(tǒng)就是對(duì)數(shù)據(jù)的處理，而對(duì)于一個(gè)擁有多個(gè)子系統(tǒng)的企業(yè)應(yīng)用系統(tǒng)而言，它的基礎(chǔ)支撐無疑就是對(duì)消息的處理。與對(duì)象不同，消息本質(zhì) 上是一種數(shù)據(jù)結(jié)構(gòu)（當(dāng)然，對(duì)象也可以看做是一種特殊的消息），它包含消費(fèi)者與服務(wù)雙方都能識(shí)別的數(shù)據(jù)，

3、這些數(shù)據(jù)需要在不同的進(jìn)程（機(jī)器）之間進(jìn)行傳遞，并 可能會(huì)被多個(gè)完全不同的客戶端消費(fèi)。在眾多分布式技術(shù)中，消息傳遞相較文件傳遞與遠(yuǎn)程過程調(diào)用（RPC）而言，似乎更勝一籌，因?yàn)樗哂懈玫钠脚_(tái)無關(guān) 性，并能夠很好地支持并發(fā)與異步調(diào)用。對(duì)于Web Service與RESTful而言，則可以看做是消息傳遞技術(shù)的一種衍生或封裝。常用的消息模式在我參與過的所有企業(yè)應(yīng)用系統(tǒng)中，無一例外地都采用（或在某些子系統(tǒng)與模塊中部分采用）了基于消息的分布式架構(gòu)。但是不同之處在于，讓我們做出架構(gòu)決策的證據(jù)卻迥然而異，這也直接影響我們所要應(yīng)用的消息模式。消息通道（Message Channel）模式我們常常運(yùn)用的消息模式是

4、Message Channel（消息通道）模式，如圖1所示。圖1 Message Channel模式消息通道作為在客戶端（消費(fèi)者，Consumer）與服務(wù)（生產(chǎn)者，Producer）之間引入的間接層，可以有效地解除二者之間的耦合。只要實(shí)現(xiàn) 規(guī)定雙方需要通信的消息格式，以及處理消息的機(jī)制與時(shí)機(jī)，就可以做到消費(fèi)者對(duì)生產(chǎn)者的“無知”。事實(shí)上，該模式可以支持多個(gè)生產(chǎn)者與消費(fèi)者。例如，我們可 以讓多個(gè)生產(chǎn)者向消息通道發(fā)送消息，因?yàn)橄M(fèi)者對(duì)生產(chǎn)者的無知性，它不必考慮究竟是哪個(gè)生產(chǎn)者發(fā)來的消息。雖然消息通道解除了生產(chǎn)者與消費(fèi)者之間的耦合，使得我們可以任意地對(duì)生產(chǎn)者與消費(fèi)者進(jìn)行擴(kuò)展，但它又同時(shí)引入了各自對(duì)消息

5、通道的依賴，因?yàn)樗鼈儽仨?知道通道資源的位置。要解除這種對(duì)通道的依賴，可以考慮引入Lookup服務(wù)來查找該通道資源。例如，在JMS中就可以通過JNDI來獲取消息通道 Queue。若要做到充分的靈活性，可以將與通道相關(guān)的信息存儲(chǔ)到配置文件中，Lookup服務(wù)首先通過讀取配置文件來獲得通道。消息通道通常以隊(duì)列的形式存在，這種先進(jìn)先出的數(shù)據(jù)結(jié)構(gòu)無疑最為適合這種處理消息的場(chǎng)景。微軟的MSMQ、IBM MQ、JBoss MQ以及開源的RabbitMQ、Apache ActiveMQ都 通過隊(duì)列實(shí)現(xiàn)了Message Channel模式。因此，在選擇運(yùn)用Message Channel模式時(shí)，更多地是要從質(zhì)量

6、屬性的層面對(duì)各種實(shí)現(xiàn)了該模式的產(chǎn)品進(jìn)行全方位的分析與權(quán)衡。例如，消息通道對(duì)并發(fā)的支持以及在性能上的表現(xiàn)；消 息通道是否充分地考慮了錯(cuò)誤處理；對(duì)消息安全的支持；以及關(guān)于消息持久化、災(zāi)備（fail over）與集群等方面的支持。因?yàn)橥ǖ纻鬟f的消息往往是一些重要的業(yè)務(wù)數(shù)據(jù)，一旦通道成為故障點(diǎn)或安全性的突破點(diǎn)，對(duì)系統(tǒng)就會(huì)造成災(zāi)難性的影響。在本文 的第二部分，我將給出一個(gè)實(shí)際案例來闡釋在進(jìn)行架構(gòu)決策時(shí)應(yīng)該考慮的架構(gòu)因素，并由此做出正確地決策。發(fā)布者-訂閱者（Publisher-Subscriber）模式一旦消息通道需要支持多個(gè)消費(fèi)者時(shí)，就可能面臨兩種模型的選擇：拉模型與推模型。拉模型是由消息的消費(fèi)者發(fā)起

7、的，主動(dòng)權(quán)把握在消費(fèi)者手中，它會(huì)根據(jù)自己的情況對(duì)生產(chǎn)者發(fā)起調(diào)用。如圖2所示：圖2 拉模型拉模型的另一種體現(xiàn)則由生產(chǎn)者在狀態(tài)發(fā)生變更時(shí)，通知消費(fèi)者其狀態(tài)發(fā)生了改變。但得到通知的消費(fèi)者卻會(huì)以回調(diào)方式，通過調(diào)用傳遞過來的消費(fèi)者對(duì)象獲取更多細(xì)節(jié)消息。在基于消息的分布式系統(tǒng)中，拉模型的消費(fèi)者通常以Batch Job的形式，根據(jù)事先設(shè)定的時(shí)間間隔，定期偵聽通道的情況。一旦發(fā)現(xiàn)有消息傳遞進(jìn)來，就會(huì)轉(zhuǎn)而將消息傳遞給真正的處理器（也可以看做是消費(fèi)者）處理消 息，執(zhí)行相關(guān)的業(yè)務(wù)。在本文第二部分介紹的醫(yī)療衛(wèi)生系統(tǒng)，正是通過引入Quartz.NET實(shí)現(xiàn)了Batch Job，完成對(duì)消息通道中消息的處理。推模型的主動(dòng)權(quán)常

8、常掌握在生產(chǎn)者手中，消費(fèi)者被動(dòng)地等待生產(chǎn)者發(fā)出的通知，這就要求生產(chǎn)者必須了解消費(fèi)者的相關(guān)信息。如圖3所示：圖3 推模型對(duì)于推模型而言，消費(fèi)者無需了解生產(chǎn)者。在生產(chǎn)者通知消費(fèi)者時(shí)，傳遞的往往是消息（或事件），而非生產(chǎn)者自身。同時(shí)，生產(chǎn)者還可以根據(jù)不同的情況，注冊(cè)不同的消費(fèi)者，又或者在封裝的通知邏輯中，根據(jù)不同的狀態(tài)變化，通知不同的消費(fèi)者。兩種模型各有優(yōu)勢(shì)。拉模型的好處在于可以進(jìn)一步解除消費(fèi)者對(duì)通道的依賴，通過后臺(tái)任務(wù)去定期訪問消息通道。壞處是需要引入一個(gè)單獨(dú)的服務(wù)進(jìn)程，以 Schedule形式執(zhí)行。而對(duì)于推模型而言，消息通道事實(shí)上會(huì)作為消費(fèi)者觀察的主體，一旦發(fā)現(xiàn)消息進(jìn)入，就會(huì)通知消費(fèi)者執(zhí)行對(duì)消息

9、的處理。無論推模型， 拉模型，對(duì)于消息對(duì)象而言，都可能采用類似Observer模式的機(jī)制，實(shí)現(xiàn)消費(fèi)者對(duì)生產(chǎn)者的訂閱，因此這種機(jī)制通常又被稱為Publisher- Subscriber模式，如圖4所示：圖4 Publisher-Subscriber模式通常情況下，發(fā)布者和訂閱者都會(huì)被注冊(cè)到用于傳播變更的基礎(chǔ)設(shè)施（即消息通道）上。發(fā)布者會(huì)主動(dòng)地了解消息通道，使其能夠?qū)⑾l(fā)送到通道中；消息通道一旦接收到消息，會(huì)主動(dòng)地調(diào)用注冊(cè)在通道中的訂閱者，進(jìn)而完成對(duì)消息內(nèi)容的消費(fèi)。對(duì)于訂閱者而言，有兩種處理消息的方式。一種是廣播機(jī)制，這時(shí)消息通道中的消息在出列的同時(shí)，還需要復(fù)制消息對(duì)象，將消息傳遞給多個(gè)訂閱者。

10、例如， 有多個(gè)子系統(tǒng)都需要獲取從CRM系統(tǒng)傳來的客戶信息，并根據(jù)傳遞過來的客戶信息，進(jìn)行相應(yīng)的處理。此時(shí)的消息通道又被稱為Propagation通道。另 一種方式則屬于搶占機(jī)制，它遵循同步方式，在同一時(shí)間只能有一個(gè)訂閱者能夠處理該消息。實(shí)現(xiàn)Publisher-Subscriber模式的消息通道會(huì)選 擇當(dāng)前空閑的唯一訂閱者，并將消息出列，并傳遞給訂閱者的消息處理方法。消息路由（Message Router）模式無論是Message Channel模式，還是Publisher-Subscriber模式，隊(duì)列在其中都扮演了舉足輕重的角色。然而，在企業(yè)應(yīng)用系統(tǒng)中，當(dāng)系統(tǒng)變得越來越 復(fù)雜時(shí)，對(duì)性能的要求

11、也會(huì)越來越高，此時(shí)對(duì)于系統(tǒng)而言，可能就需要支持同時(shí)部署多個(gè)隊(duì)列，并可能要求分布式部署不同的隊(duì)列。這些隊(duì)列可以根據(jù)定義接收不同 的消息，例如訂單處理的消息，日志信息，查詢?nèi)蝿?wù)消息等。這時(shí)，對(duì)于消息的生產(chǎn)者和消費(fèi)者而言，并不適宜承擔(dān)決定消息傳遞路徑的職責(zé)。事實(shí)上，根據(jù)S單一 職責(zé)原則，這種職責(zé)分配也是不合理的，它既不利于業(yè)務(wù)邏輯的重用，也會(huì)造成生產(chǎn)者、消費(fèi)者與消息隊(duì)列之間的耦合，從而影響系統(tǒng)的擴(kuò)展。既然這三種對(duì)象（組件）都不宜承擔(dān)這樣的職責(zé)，就有必要引入一個(gè)新的對(duì)象專門負(fù)責(zé)傳遞路徑選擇的功能，這就是所謂的Message Router模式，如圖5所示：圖5 Message Router模式通過消息路

12、由，我們可以配置路由規(guī)則指定消息傳遞的路徑，以及指定具體的消費(fèi)者消費(fèi)對(duì)應(yīng)的生產(chǎn)者。例如指定路由的關(guān)鍵字，并由它來綁定具體的隊(duì)列與 指定的生產(chǎn)者（或消費(fèi)者）。路由的支持提供了消息傳遞與處理的靈活性，也有利于提高整個(gè)系統(tǒng)的消息處理能力。同時(shí)，路由對(duì)象有效地封裝了尋找與匹配消息路 徑的邏輯，就好似一個(gè)調(diào)停者（Meditator），負(fù)責(zé)協(xié)調(diào)消息、隊(duì)列與路徑尋址之間關(guān)系。除了以上的模式之外，Messaging模式提供了一個(gè)通信基礎(chǔ)架構(gòu)，使得我們可以將獨(dú)立開發(fā)的服務(wù)整合到一個(gè)完整的系統(tǒng)中。 Message Translator模式則完成對(duì)消息的解析，使得不同的消息通道能夠接收和識(shí)別不同格式的消息。而且通過

13、引入這樣的對(duì)象，也能夠很好地避免出現(xiàn)盤根錯(cuò) 節(jié)，彼此依賴的多個(gè)服務(wù)。Message Bus模式可以為企業(yè)提供一個(gè)面向服務(wù)的體系架構(gòu)。它可以完成對(duì)消息的傳遞，對(duì)服務(wù)的適配與協(xié)調(diào)管理，并要求這些服務(wù)以統(tǒng)一的方式完成協(xié)作。2、消息模式的應(yīng)用場(chǎng)景基于消息的分布式架構(gòu)總是圍繞著消息來做文章。例如可以將消息封裝為對(duì)象，或者指定消息的規(guī)范例如SOAP，或者對(duì)實(shí)體對(duì)象的序列化與反序列化。這些方式的目的只有一個(gè)，就是將消息設(shè)計(jì)為生產(chǎn)者和消費(fèi)者都能夠明白的格式，并能通過消息通道進(jìn)行傳遞。場(chǎng)景一：基于消息的統(tǒng)一服務(wù)架構(gòu)在制造工業(yè)的CIMS系統(tǒng)中，我們嘗試將各種業(yè)務(wù)以服務(wù)的形式公開給客戶端的調(diào)用者，例如定義這樣的接口

14、： public interface IService IMessage Execute(IMessage aMessage); void SendRequest(IMessage aMessage); 之所以能夠設(shè)計(jì)這樣的服務(wù)，原因在于我們對(duì)業(yè)務(wù)信息進(jìn)行了高度的抽象，以消息的形式在服務(wù)之間傳遞。此時(shí)的消息其實(shí)是生產(chǎn)者與消費(fèi)者之間的契約或接口，只要遵循這樣的契約，按照規(guī)定的格式對(duì)消息進(jìn)行轉(zhuǎn)換與抽取，就能很好地支持系統(tǒng)的分布式處理。在這個(gè)CIMS系統(tǒng)中，我們將消息劃分為ID，Name和Body，通過定義如下的接口方法，可以獲得消息主體的相關(guān)屬性： public interface IMessag

15、e:ICloneable string MessageID get; set; string MessageName() get; set; IMessageItemSequence CreateMessageBody(); IMessageItemSequence GetMessageBody(); 消息主體類Message實(shí)現(xiàn)了IMessage接口。在該類中，消息體Body為IMessageItemSequence類型。這個(gè)類型用于獲取和設(shè)置消息的內(nèi)容：Value和Item： public interface IItemValueSetting string getSubValue(str

16、ing name); void setSubValue(string name, string value); public interface IMessageItemSequence:IItemValueSetting, ICloneable IMessageItem GetMessageItem(string aName); IMessageItem CreateMessageItem(string aName); Value為字符串類型，它利用了HashTable存儲(chǔ)Key和Value的鍵值對(duì)。Item則為IMessageItem類型，在IMessageItemSequence的實(shí)現(xiàn)類

17、中，同樣利用了HashTable存儲(chǔ)Key和Item的鍵值對(duì)。IMessageItem支持消息體的嵌套。它包含了兩部分：SubValue和SubItem。實(shí)現(xiàn)的方式和IMessageItemSequence相似。通過定義這樣的嵌套結(jié)構(gòu)，使得消息的擴(kuò)展成為可能。一般的消息結(jié)構(gòu)如下所示：各個(gè)消息對(duì)象之間的關(guān)系如圖6所示：圖6 消息對(duì)象之間的關(guān)系在實(shí)現(xiàn)服務(wù)進(jìn)程通信之前，我們必須定義好各個(gè)服務(wù)或各個(gè)業(yè)務(wù)的消息格式。通過消息體的方法在服務(wù)的一端設(shè)置消息的值，然后發(fā)送，并在服務(wù)的另一端獲得這些值。例如發(fā)送消息端定義如下的消息體：IMessageFactory factory = new MessageFa

18、ctory(); IMessage message = factory.CreateMessage(); message.SetMessageName("service1"); IMessageItemSequence body = message.CreateMessageBody(); body.SetSubValue("subname1","subvalue1"); body.SetSubValue("subname2","subvalue2"); IMessageItem item1 =

19、 body.CreateMessageItem(”item1”); item1.SetSubValue("subsubname11","subsubvalue11"); item1.SetSubValue("subsubname12","subsubvalue12"); /Send Request Message MyServiceClient service = new MyServiceClient("Client"); service.SendRequest(message);我們?cè)诳蛻舳?/p>

20、引入了一個(gè)ServiceLocator對(duì)象，它通過MessageQueueListener對(duì)消息隊(duì)列進(jìn)行偵聽，一旦接收到消息，就獲取該消息中的name去定位它所對(duì)應(yīng)的服務(wù)，然后調(diào)用服務(wù)的Execute(aMessage)方法，執(zhí)行相關(guān)的業(yè)務(wù)。ServiceLocator承擔(dān)的定位職責(zé)其實(shí)是對(duì)存儲(chǔ)在ServiceContainer容器中的服務(wù)進(jìn)行查詢。 ServiceContainer容器可以讀取配置文件，在啟動(dòng)服務(wù)的時(shí)候初始化所有的分布式服務(wù)（注意，這些服務(wù)都是無狀態(tài)的），并對(duì)這些服務(wù)進(jìn)行管 理。它封裝了服務(wù)的基本信息，諸如服務(wù)所在的位置，服務(wù)的部署方式等，從而避免服務(wù)的調(diào)用者直接依賴于服務(wù)的

21、細(xì)節(jié)，既減輕了調(diào)用者的負(fù)擔(dān)，還能夠較好地實(shí) 現(xiàn)服務(wù)的擴(kuò)展與遷移。在這個(gè)系統(tǒng)中，我們主要引入了Messaging模式，通過定義的IMessage接口，使得我們更好地對(duì)服務(wù)進(jìn)行抽象，并以一種扁平的格式存儲(chǔ)數(shù) 據(jù)信息，從而解除服務(wù)之間的耦合。只要各個(gè)服務(wù)就共用的消息格式達(dá)成一致，請(qǐng)求者就可以不依賴于接收者的具體接口。通過引入的Message對(duì)象，我們就 可以建立一種在行業(yè)中通用的消息模型與分布式服務(wù)模型。事實(shí)上，基于這樣的一個(gè)框架與平臺(tái)，在對(duì)制造行業(yè)的業(yè)務(wù)進(jìn)行開發(fā)時(shí)，開發(fā)人員最主要的活動(dòng)是與領(lǐng)域 專家就各種業(yè)務(wù)的消息格式進(jìn)行討論，這樣一種面向領(lǐng)域的消息語言，很好地掃清了技術(shù)人員與業(yè)務(wù)人員的溝通障礙；

22、同時(shí)在各個(gè)子系統(tǒng)之間，我們也只需要維護(hù)服 務(wù)間相互傳遞的消息接口表。每個(gè)服務(wù)的實(shí)現(xiàn)都是完全隔離的，有效地做到了對(duì)業(yè)務(wù)知識(shí)與基礎(chǔ)設(shè)施的合理封裝與隔離。對(duì)于消息的格式和內(nèi)容，我們考慮引入了Message Translator模式，負(fù)責(zé)對(duì)前面定義的消息結(jié)構(gòu)進(jìn)行翻譯和解析。為了進(jìn)一步減輕開發(fā)人員的負(fù)擔(dān)，我們還可以基于該平臺(tái)搭建一個(gè)消息-對(duì)象-關(guān)系的映 射框架，引入實(shí)體引擎（Entity Engine）將消息轉(zhuǎn)換為領(lǐng)域?qū)嶓w，使得服務(wù)的開發(fā)者能夠以完全面向?qū)ο蟮乃枷腴_發(fā)各個(gè)服務(wù)組件，并通過調(diào)用持久層實(shí)現(xiàn)消息數(shù)據(jù)的持久化。同時(shí)，利用消 息總線（此時(shí)的消息總線可以看做是各個(gè)服務(wù)組件的連接器）連接不同的服務(wù)，并

23、允許異步地傳遞消息，對(duì)消息進(jìn)行編碼。這樣一個(gè)基于消息的分布式架構(gòu)如圖7所 示：圖7 基于Message Bus的CIMS分布式架構(gòu)場(chǎng)景二：消息中間件的架構(gòu)決策在一個(gè)醫(yī)療衛(wèi)生系統(tǒng)中，我們面臨了客戶對(duì)系統(tǒng)性能/可用性的非功能需求。在我們最初啟動(dòng)該項(xiàng)目時(shí)，客戶就表達(dá)了對(duì)性能與可用性的特別關(guān)注?？蛻粝Ｍ?最終用戶在進(jìn)行復(fù)雜的替換刪除操作時(shí)，能夠具有很好的用戶體驗(yàn)，簡(jiǎn)言之，就是希望能夠快速地得到操作的響應(yīng)。問題在于這樣的替換刪除操作需要處理比較復(fù)雜 的業(yè)務(wù)邏輯，同時(shí)牽涉到的關(guān)聯(lián)數(shù)據(jù)量非常大，整個(gè)操作若需完成，最壞情況下可能需要幾分鐘的時(shí)間。我們可以通過引入緩存、索引、分頁等多種方式對(duì)數(shù)據(jù)庫操 作進(jìn)行性能

24、調(diào)優(yōu)，但整個(gè)操作的耗時(shí)始終無法達(dá)到客戶的要求。由于該系統(tǒng)是在一個(gè)遺留系統(tǒng)的基礎(chǔ)上開發(fā)，如果要引入Map-Reduce來處理這些操作，以 滿足質(zhì)量需求，則對(duì)架構(gòu)的影響太大，且不能很好地重用之前系統(tǒng)的某些組件。顯然，付出的成本與收益并不成正比。通過對(duì)需求進(jìn)行分析，我們注意到最終客戶并不需要實(shí)時(shí)獲得結(jié)果，只要能夠保證最終結(jié)果的一致性和完整性即可。關(guān)鍵在于就用戶體驗(yàn)而言，他們不希望經(jīng)歷漫長(zhǎng)的等待，然后再通知他們操作究竟是成功還是失敗。這是一個(gè)典型需要通過后臺(tái)任務(wù)進(jìn)行異步處理的場(chǎng)景。在企業(yè)應(yīng)用系統(tǒng)中，我們常常會(huì)遭遇這樣的場(chǎng)景。我們?cè)?jīng)在一個(gè)金融系統(tǒng)中嘗試通過自己編寫任務(wù)的方式來控制后臺(tái)線程的并發(fā)訪問，并

25、完成對(duì)任務(wù)的調(diào)度。事實(shí)證明，這樣的設(shè)計(jì)并非行之有效。對(duì)于這種典型的異步處理來說，基于消息傳遞的架構(gòu)模式才是解決這一問題的最佳辦法。因?yàn)橄⒅虚g件的逐步成熟，對(duì)于這一問題的架構(gòu)設(shè)計(jì)，已經(jīng)由原來對(duì)設(shè)計(jì)實(shí)現(xiàn)的關(guān)注轉(zhuǎn)為如何進(jìn)行產(chǎn)品選型和技術(shù)決策。通過分析業(yè)務(wù)場(chǎng)景以及客戶性質(zhì)，我們發(fā)現(xiàn)該業(yè)務(wù)場(chǎng)景具有如下特征：· 在一些特定情形下，可能會(huì)集中發(fā)生批量的替換刪除操作，使得操作的并發(fā)量達(dá)到高峰；例如FDA要求召回一些違規(guī)藥品時(shí)，就需要?jiǎng)h除藥品庫中該藥品的信息；· 操作結(jié)果不要求實(shí)時(shí)性，但需要保證操作的可靠性，不能因?yàn)楫惓Ｊ《鴮?dǎo)致某些操作無法進(jìn)行；· 自動(dòng)操作的過程是不可逆轉(zhuǎn)的，

26、因此需要記錄操作歷史；· 基于性能考慮，大多數(shù)操作需要調(diào)用數(shù)據(jù)庫的存儲(chǔ)過程；· 操作的數(shù)據(jù)需要具備一定的安全性，避免被非法用戶對(duì)數(shù)據(jù)造成破壞；· 與操作相關(guān)的功能以組件形式封裝，保證組件的可重用性、可擴(kuò)展性與可測(cè)試性；· 數(shù)據(jù)量可能隨著最終用戶的增多而逐漸增大；針對(duì)如上的業(yè)務(wù)需求，我們決定從以下幾個(gè)方面對(duì)各種技術(shù)方案進(jìn)行橫向的比較與考量。· 并發(fā)：選擇的消息隊(duì)列一定要很好地支持用戶訪問的并發(fā)性；· 安全：消息隊(duì)列是否提供了足夠的安全機(jī)制；· 性能伸縮：不能讓消息隊(duì)列成為整個(gè)系統(tǒng)的單一性能瓶頸；· 部署：盡可能讓消息

27、隊(duì)列的部署更為容易；· 災(zāi)備：不能因?yàn)橐馔獾腻e(cuò)誤、故障或其他因素導(dǎo)致處理數(shù)據(jù)的丟失；· API易用性：處理消息的API必須足夠簡(jiǎn)單、并能夠很好地支持測(cè)試與擴(kuò)展；我們先后考察了MSMQ、Resque、ActiveMQ和RabbitMQ，通過查詢相關(guān)資料，以及編寫Spike代碼驗(yàn)證相關(guān)質(zhì)量，我們最終選擇了RabbitMQ。我們選擇放棄MSMQ，是因?yàn)樗鼑?yán)重依賴Windows操作系統(tǒng)；它雖然提供了易用的GUI方便管理人員對(duì)其進(jìn)行安裝和部署，但若要編寫自動(dòng)化部署 腳本，卻非常困難。同時(shí)，MSMQ的隊(duì)列容量不能查過4M字節(jié)，這也是我們無法接收的。Resque的問題是目前僅支持Ruby

28、的客戶端調(diào)用，不能很好地 與.NET平臺(tái)集成。此外，Resque對(duì)消息持久化的處理方式是寫入到Redis中，因而需要在已有RDBMS的前提下，引入新的Storage。我們 比較傾心于ActiveMQ與RabbitMQ，但通過編寫測(cè)試代碼，采用循環(huán)發(fā)送大數(shù)據(jù)消息以驗(yàn)證消息中間件的性能與穩(wěn)定性時(shí)，我們發(fā)現(xiàn) ActiveMQ的表現(xiàn)并不太讓人滿意。至少，在我們的詢證調(diào)研過程中，ActiveMQ會(huì)因?yàn)轭l繁發(fā)送大數(shù)據(jù)消息而偶爾出現(xiàn)崩潰的情況。相對(duì)而 言，RabbitMQ在各個(gè)方面都比較適合我們的架構(gòu)要求。例如在災(zāi)備與穩(wěn)定性方面，RabbitMQ提供了可持久化的隊(duì)列，能夠在隊(duì)列服務(wù)崩潰的時(shí)候，將未處理的消息

29、持久化到磁盤上。為了避免因?yàn)榘l(fā)送消息 到寫入消息之間的延遲導(dǎo)致信息丟失，RabbitMQ引入了Publisher Confirm機(jī)制以確保消息被真正地寫入到磁盤中。它對(duì)Cluster的支持提供了Active/Passive與Active/Active兩種模 式。例如，在Active/Passive模式下，一旦一個(gè)節(jié)點(diǎn)失敗，Passive節(jié)點(diǎn)就會(huì)馬上被激活，并迅速替代失敗的Active節(jié)點(diǎn)，承擔(dān)起消息 傳遞的職責(zé)。如圖8所示：圖8 Active/Passive Cluster在并發(fā)處理方面，RabbitMQ本身是基于erlang編寫的消息中間件，作為一門面向并發(fā)處理的編程語言，erlang對(duì)并發(fā)

30、處理的天生優(yōu)勢(shì)使 得我們對(duì)RabbitMQ的并發(fā)特性抱有信心。RabbitMQ可以非常容易地部署到Windows、Linux等操作系統(tǒng)下，同時(shí)，它也可以很好地部署 到服務(wù)器集群中。它的隊(duì)列容量是沒有限制的（取決于安裝RabbitMQ的磁盤容量），發(fā)送與接收信息的性能表現(xiàn)也非常好。RabbitMQ提供了 Java、.NET、Erlang以及C語言的客戶端API，調(diào)用非常簡(jiǎn)單，并且不會(huì)給整個(gè)系統(tǒng)引入太多第三方庫的依賴。 例如.NET客戶端只需要依賴一個(gè)程序集。即使我們選擇了RabbitMQ，但仍有必要對(duì)系統(tǒng)與具體的消息中間件進(jìn)行解耦，這就要求我們對(duì)消息的生產(chǎn)者與消費(fèi)者進(jìn)行抽象，例如定義如下的接口：

31、public interface IQueueSubscriber void ListenTo<T>(string queueName, Action<T> action); void ListenTo<T>(string queueName, Predicate<T> messageProcessedSuccessfully); void ListenTo<T>(string queueName, Predicate<T> messageProcessedSuccessfully, bool requeueFailedM

32、essages); public interface IQueueProvider T Pop<T>(string queueName); T PopAndAwaitAcknowledgement<T>(string queueName, Predicate<T> messageProcessedSuccessfully); T PopAndAwaitAcknowledgement<T>(string queueName, Predicate<T> messageProcessedSuccessfully, bool requeueF

33、ailedMessages); void Push(FunctionalArea functionalArea, string routingKey, object payload); 在這兩個(gè)接口的實(shí)現(xiàn)類中，我們封裝了RabbitMQ的調(diào)用類，例如：public class RabbitMQSubscriber : IQueueSubscriber public void ListenTo<T>(string queueName, Action<T> action) using (IConnection connection = _factory.OpenConnec

34、tion() using (IModel channel = connection.CreateModel() var consumer = new QueueingBasicConsumer(channel); string consumerTag = channel.BasicConsume(queueName, AcknowledgeImmediately, consumer); var response = (BasicDeliverEventArgs) (); var serializer = new JavaScriptSerializer(); string json = Enc

35、oding.UTF8.GetString(response.Body); var message = serializer.Deserialize<T>(json); action(message); public class RabbitMQProvider : IQueueProvider public T Pop<T>(string queueName) var returnVal = default(T); const bool acknowledgeImmediately = true; using (var connection = _factory.Ope

36、nConnection() using (var channel = connection.CreateModel() var response = channel.BasicGet(queueName, acknowledgeImmediately); if (response != null) var serializer = new JavaScriptSerializer(); var json = Encoding.UTF8.GetString(response.Body); returnVal = serializer.Deserialize<T>(json); ret

37、urn returnVal; 我們用Quartz.Net來實(shí)現(xiàn)Batch Job。通過定義一個(gè)實(shí)現(xiàn)了IStatefulJob接口的Job類，在Execute()方法中完成對(duì)隊(duì)列的偵聽。Job中 RabbitMQSubscriber類的ListenTo()方法會(huì)調(diào)用Queue的Dequeue()方法，當(dāng)接收的消息到達(dá)隊(duì)列時(shí)，Job會(huì)偵聽到 消息達(dá)到的事件，然后以同步的方式使得消息彈出隊(duì)列，并將消息作為參數(shù)傳遞給Action委托。因此，在Batch Job的Execute()方法中，可以定義消息處理的方法，并調(diào)用RabbitMQSubscriber類的ListenTo()方法，如下所示（注 意，這

38、里傳遞的消息事實(shí)上是Job的Id）： public void Execute(JobExecutionContext context) string queueName = queueConfigurer.GetQueueProviders().Queue.Name;try queueSubscriber.ListenTo<MyJob>( queueName, job => request.MakeRequest(); catch(Exception err) Log.WarnFormat("Unexpected exception while processing

39、 queue '0', Details: 1", queueName, err); 隊(duì)列的相關(guān)信息例如隊(duì)列名都存儲(chǔ)在配置文件中。Execute()方法調(diào)用了request對(duì)象的MakeRequest()方法，并將獲得的消息（即JobId）傳遞給該方法。它會(huì)根據(jù)JobId到數(shù)據(jù)庫中查詢?cè)揓ob對(duì)應(yīng)的信息，并執(zhí)行真正的業(yè)務(wù)處理。在對(duì)基于消息處理的架構(gòu)進(jìn)行決策時(shí)，除了前面提到的考慮因素外，還需要就許多設(shè)計(jì)細(xì)節(jié)進(jìn)行多方位的判斷與權(quán)衡。例如針對(duì)Job的執(zhí)行以及隊(duì)列的管理，就需要考慮如下因素：· 對(duì)Queue中Job狀態(tài)的監(jiān)控與查詢；· 對(duì)Job優(yōu)先級(jí)的管理；· 能否取消或終止執(zhí)行時(shí)間過長(zhǎng)的Job；· 是否能夠設(shè)定Job的執(zhí)行時(shí)間；· 是否能夠設(shè)定Poll的間隔時(shí)間；· 能否跨機(jī)器分布式的放入Job；· 對(duì)失敗Job的處