消息中間件
1.消息中間件的價值
1.1 透過示例看消息中間件對應用的解耦
1.1.1.通過服務調用讓其他系統感知事件發生的方式
假設我們要做一個用戶登錄系統，其中需要支持的一個功能是，用戶登錄成功 后發送一條短信到用戶的手機，算是一個用戶安全的選項，如圖6-2所示。
然后，我們需要把用戶登錄的信息（時間、IP、用戶名等數據）傳給我們的安全 系統，安全系統會進行安全策略相關的處理和判斷。此時的結構會變成圖6-3所示的 樣子。
這樣看起來還好，但是如果再增加一些登錄成功后需要被調用的系統呢？如圖 6-4所示。
這會讓登錄系統變得非常復雜。每增加一個在登錄成功后需要被調用的系統， 就需要修改登錄系統來進行相關的調用。優雅一點的實現是把這個登錄成功后的服 務調用變為一種可擴展的配置，甚至可以動態生效，但這只能降低變更時的開發和 部署成本，並沒有降低復雜性。登錄系統要被迫依賴非常多的系統。
1.1.2.通過引入消息中間件解耦服務調用
我們思考一下，如果從登錄系統的角度來看，這些系統是登錄系統必須依賴的 嗎？答案是否定的，.登錄系統只需要驗證用戶名和密碼的合法性（也許還會包含驗證碼等登錄驗證合法性的必需要素），所以登錄系統必須依賴的是能夠提供用戶名、 密碼的系統，而圖6-4中的系統其實都不是登錄系統必須依賴的系統。相反，這些系統是必須依賴登錄系統的，因為它們都關心登錄是否成功這件事情。
在這樣的場景中，我們需要通過消息中間件把上面的結構解耦，上面結構中的服務調用將會被固定格式的消息的傳遞所取代。登錄系統負責向消息中間件發送消 息，而其他的系統則向消息中間件來訂閱這個消息，然后完成自己的工作，如圖6-5 所示。
通過消息中間件解耦，登錄系統就不用關心到底有多少個系統需要知道登錄成 功這件事了，也不用關心如何通知它們，只需要把登錄成功這件事轉化為一個消息 發送到消息中間件就可以了。這樣，需要了解登錄成功這件事的系統自己去消息中 間件訂閱就行了。並且各個系統之間也是互不影響的。
這里需要注意一點，就是當登錄成功時需要向消息中間件發送一個消息，那么 我們必須保證這個消息發送到了消息中間件，否則依賴這個消息的系統就無法工作 了。這個問題有一個不太優雅的解決方式，如圖6-6所示。
圖6-6所示的思路是，我們在數據庫中記錄狀態，然后讓用到這個狀態的系統自己來查。這個記錄狀態的數據庫是操作中一定會依賴的數據庫，如果它出問題就會導致對狀態的記錄不成功，業務操作也就不會成功了。圖6-6所示是把狀態記錄在用戶庫的用戶記錄上了。不過這個例子是有一個小問題的，就是如果用戶讀信息時數據庫正常，這時就能完成密碼的驗證，但是如果去記錄狀態時數據庫不可用， 那就還是有問題的（后面我們會看到如何解決）。如果這里只是對數據庫的寫操作 的話，那就沒有問題了，例如修改用戶信息的操作，那么是可以在一個SQL中完成用戶信息的更改並設置要發送短信這個狀態，這樣可以保證操作本身和狀態更新的原子性。
對於需要感知狀態的應用來說，需要定時輪詢數據庫以查看狀態，並且在做完操作后，需要更改狀態從而使得下次就不用再處理了。
可以說這是一個能解決問題的work around方法，實現也比較簡單。不過也存在以下幾個問題：
•增加了業務數據庫的負擔。一個狀態字段所占的空間還可以接受，但是這個數據庫需要被其他系統持續地定時輪詢，並且進行更新，這就大大增加了數據庫的負擔。
•依賴的復雜和不安全。該方案使得發送短信的服務要依賴業務數據庫，這導致依賴復雜並且不合理，另外，發送短信的服務對數據庫記錄有修改的權限， 這也不安全。
•擴展性不好。對於前面的多個需要在業務動作成功后來做后續工作的系統， 如果把該方式用於這樣的系統的話，我們就需要增加很多個字段，或者使這 些字段變得可共享又相互不能影響。並且會增加大量的定時對業務數據庫的 輪詢請求。
對於這些問題，我們也期望通過消息中間件來解決。
2.互聯網時代的消息中間件
在開始介紹互聯網時代的消息中間件前，我們必須講一下JMS.JMS是Java Message Service的縮寫，它是Java EE (企業版Java)中的一個關於消息的規范，而Hometq、 ActiveMQ等產品是對這個規范的實現。如果是企業內部或者一些小型的系統，直接使 用JMS的實現產品是一個經濟的選擇，而在大型系統中有一些場景不適合使用JMS。
在大型互聯網中，我們采用消息中間件可以進行應用之間的解耦以及操作的異步，這是消息中間件的兩個最基礎的特點，也正是我們需要的。在此基礎上，我們着重思考的是消息的順序保證、擴展性、可靠性、業務操作與消息發送一致性，以 及多集群訂閱者等方面的問題，這些內容會在后面的小節中呈現給讀者。我們從上 一節提到的保證消息一定被處理開始介紹。
2.1 如何解決消息發送一致性
2.1.1消息發送一致性的定義
首先，我們需要弄清楚消息發送一致性究竟是什么。消息發送一致性是指產生消息的業務動作與消息發送的一致，就是說，如果業務操作成功了，那么由這個操作產生的消息一定要發送出去，否則就丟失消息了。而另一方面，如果這個業務行為沒有發生或者失敗，那么就不應該把消息發出去。
2.1.2消息發送一致性很難保證嗎
如果要寫處理業務邏輯的代碼和發送消息的代碼，該怎么寫呢？
下面是一段偽代碼，是在某些實踐中的用法。從中可以看到以下兩個問題。
void fool (){
//業務操作
//例如寫數據庫，調用服務等
//發送消息 }
•業務操作在前，發送消息在后，如果業務失敗了還好（當然業務自己不覺得好），如果成功了，而這時這個應用出問題，那么消息就發不出去了。
•如果業務成功，應用也沒有掛掉，但是這時消息系統掛掉了，也會導致消息發不出去。 我們來看另外一種做法，偽代碼如下：
void fool () {
//發送消息 //業務操作
//例如寫數據庫，調用服務等 }
這種方式更不可靠，在業務還沒有做時消息就發出了。
在具體的工程實踐中，第一種做法丟失消息的比例相對是很低的。當然，對於要求必須保證一致性的場景，上面的兩種方案都不能接受。
2.1.3大家熟知的JMS有辦法嗎
使用JMS可以實現消息發送一致性嗎？我們來看看JMS發送消息的部分。首先看看JMS中幾個比較重要的要素。
- Destination,是指消息所走通道的目標定義，也就是用來定義消息從發送端 出后要走的通道，而不是最終接收方。Destination屬於管理類的對象。
- ConnectionFactory，從名字就能看出來，是指用於創建連接的對象 ConnectionFactory屬於管理類的對象。
- Connection,連接接口，所負責的重要工作是創建Session。
- Session,會話接口，這是一個非常重要的對象，消息的發送者、接收者以及 消息對象本身，都是由這個會話對象創建的。
- MessageConsumer,消息的消費者，也就是訂閱消息並處理消息的對象。
- MessageProducer,消息的生產者，就是用來發送消息的對象。
- XXXMessage,是指各種類型的消息對象，包括BytesMessage、MapMessage、 ObjectMessage、StreamMessage 和 TextMessage 5 種。
在JMS消息模型中，有Queue和Topic (在后面會詳細介紹）之分，所以，前面的 Destination、ConnectionFactory、Connection、Session、MessageConsumer、 MessageProducer都有對應的子接口。表6-1顯示了前面各要素在Queue模型（PTP Domain)和 Topic 模型（Pub/Sub Domain)下的對應關系。
此外，在JMS的API中，我們看到很多以XA開頭的接口，它們其實就是支持 XA協議的接口，它們與表6-1中各要素的對應關系如表6-2所示。
可以看到，XA 系列的接口集中在 ConnectionFactory、Connection 和 Session 上， 而 MessageProducer、QueueSender、TopicPublisher、MessageConsumer、QueueReceiver 和TopicSubscriber則沒有對應的XA對象。這是因為事務的控制是在Session層面上 的，而 Session 是通過 Connection 創建的，Connection 是通過 ConnectionFactory 創建 的，所以，這三個接口需要有XA系列對應的接口的定義。Session、Connection、 ConnectionFactory在Queue模型和Topic模型下對應的各個接口也存在相應的XA系 列的對應接口。
下面展示了消息最重要的要素（消息、發送者、接收者）與幾個基本元素之間 的關系。
ConnectionFactory->Connection->Session->Message Destination + Session-^ MessageProducer Destination + Sessoin-^ MessageConsumer
在JMS中，如果不使用XA系列的接口實現，那么我們就無法直接得到發送消息給消息中間件及業務操作這兩個事情的事務保證，而JMS中定義的XA系列的接口就是為了實現分布式事務的支持（發送消息和業務操作很難做在一個本地事務中， 后面會講到一些變通的做法)。但是這會帶來如下問題。
•引入了分布式事務，這會帶來一些開銷並增加復雜性。
•對於業務操作有限制，要求業務操作的資源必須支持XA協議，才能夠與發送消息一起來做分布式事務。這會成為一個限制，因為並不是所有需要與發送消息一起做成分布式事務的業務操作都支持XA協議。
2.1.4 有其他的辦法嗎
從上節可以看到，jMS是可以解決消息發送一致性的問題的，但是存在一 些限制並且成本相對較高。那么，我們有沒有其他的辦法呢？
我們來思考一下要解決的問題，我們希望保證業務操作與發送相關消息的動作是一致的，而前面的簡單方案不能完全保證，但是出現問題的概率並不大，所以， 我們希望找到一種解決方案，這種方案對正常流程的影響要盡可能小，而在有問題 的場景能解決問題。
從這個方面看，即便可以做到業務操作都是支持XA的，如果采用這樣的方式引人兩階段提交的話，那么還是把方案做得有些重了。
針對這個問題，我們可以用圖6-7所示的方案來解決，流程介紹如下。
(1) 業務處理應用首先把消息發給消息中間件，標記消息的狀態為待處理。
(2) 消息中間件收到消息后，把消息存儲在消息存儲中，並不投遞該消息。
(3) 消息中間件返回消息處理的結果（僅是入庫的結果），結果是成功或者失敗。
(4) 業務方收到消息中間件返回的結果並進行處理：
a) 如果收到的結果是失敗，那么就放棄業務處理，結束。
b) 如果收到的結果是成功，則進行業務自身的操作。
(5) 業務操作完成，把業務操作的結果發送給消息中間件。
(6) 消息中間件收到業務操作結果，根據結果進行處理：
a) 如果業務失敗，則刪除消息存儲中的消息，結束。
b) 如果業務成功，則更新消息存儲中的消息狀態為可發送，並且進行調度，進行消息的投遞。
這就是整個流程。在這里讀者一定會有一個疑問，即在最簡單的版本中，我們 只有業務操作和發消息兩步，仍然會可能產生很多異常，那么現在這個過程的步驟 更多，產生異常的可能點更多，是如何能夠保證業務操作和發送消息到消息中間件 是一致的呢？
我們對每一個步驟可能產生的異常情況來進行分析。
(1) 業務應用發消息給消息中間件。如果這一步失敗了，無論是網絡的原因還 是消息中間件的原因，或是業務應用自身的原因，我們都會看到業務操作沒有做， 消息也沒有被存儲在消息中間件中，業務操作和消息的狀態是一樣的，沒有問題。
(2) 消息中間件把消息入庫。如果這一步失敗，無論是消息存儲有問題，還是 消息中間件收到業務消息后有問題，或是網絡問題，可能造成的結果有兩個。一個 是消息中間件失效，那么業務應用是收不到消息中間件的返回結果的；二是消息中 間件插人消息失敗，並且有能力返回結果給應用，這時消息存儲中都沒有消息。
(3) 業務應用接收消息中間件返回結果異常。這里出現異常的原因可能是網絡、 消息中間件的問題，也可能是業務應用自身的問題。如果業務應用自身沒問題，那 么業務應用並不知道消息在消息中間件的處理結果，就會按照消息發送失敗來處理， 如果這時消息在消息中間件那里入庫成功的話，就會造成不一致。如果是業務應用 有問題，那么如果消息在消息中間件中處理成功的話，也就會造成不一致了；如果 未處理成功，則還是一致的。
(4) 業務應用進行業務操作。這一步不會產生太大問題。
(5) 業務應用發送業務操作結果給消息中間件。如果這一步出現問題，那么消 息中間件將不知道該如何處理已經存儲在消息存儲中的消息，可能會造成不一致。
(6) 消息中間件更新消息狀態。如果這一步出現問題，與上一步所造成的結果 是類似的。
從上面的分析可以看出，需要了解的兩個主要的控制狀態和流程的節點就是業務應用和消息中間件，我們可以分別從業務應用和消息中間件的視角來梳理一下， 如表6-3和表6-4所示。
從上面的梳理和分析可以看到，對於各種異常情況我們遇到的狀態有如下三種:
- 業務操作未進行，消息未入存儲。
- 業務操作未進行，消息存人存儲，狀態為待處理。
- 業務操作成功，消息存人存儲，狀態為待處理。
這三種情況中，第一種情況不需要進行額外的處理，因為本身就是一致的；第 二種和第三種都需要了解業務操作的結果，然后來處理已經在消息存儲中、狀態是待處理的消息。
那么如何了解業務操作的結果呢？
圖6-8展示了這個過程。由消息中間件主動詢問業務應用，獲取待處理消息所對應的業務操作的結果，然后業務應用需要對業務操作的結果進行檢查，並且把結果發送給消息中間件（業務處理結果有失敗、成功、等待三種，等待是多出來的一種狀態，代表業務操作還在處理中），然后消息中間件根據這個處理結果，更新消息狀態。可以說這是發送消息的一個反向的流程。
同樣，這個流程也會出現很多異常。不過這個4步的流程就是為了確認業務處理操作結果，真正的操作只是根據業務處理結果來更改消息的狀態，所以，前面3 步都與查詢相關，如果失敗就失敗了，而最后一步的更新狀態如果失敗了，那么就定時重復這個反向流程，重復查詢就可以了。
發送消息的正向流程和檢查業務操作結果的反向流程合起來，就是解決業務操作與發送消息一致性的方案。在大多數的情況下，反向流程是不需要工作的。我們來看看正向流程是否帶來了額外的負擔，對比如表6-5所示。
從上面的對比可以看到，解決一致性的方案是只增加了一次網絡操作和一次更新存儲中消息狀態的操作，就是第5步和第6步兩步。而前面4步和傳統方式所做的事情都一樣，只是順序有所不同。所以，整體上帶來的額外開銷並不大，而且還 有可優化的點。
接着來看一下使用方式。可以看到解決一致性的方案中，在業務應用那里是有一個固化的流程的，可以提供一個封裝來方便業務應用的使用，偽代碼如下。
Result postMessage(Message, PostMessageCallback){
//發送消息給消息中間件 //獲取返回結果 //如果失敗，返回失敗 //進行業務操作
//獲取業務操作結果 / /發送業務操作結果給消息中間件 //返回處理結果 }
可以看到，我們可以把實現邏輯封裝在一個調用中，然后把業務的操作包裝成 一個對象傳進來，然后整個流程就可以控制在這個方法中了。當然，除了發送一致性的消息之外，也應該提供一個傳統的發送消息的接口， 也就是不支持發送一致性的發送接口。此外，為了適應其他的場景（例如與現有的事務處理流程結合等），也會提供獨立的接口，就會把這個流程的控制權交給業務應用自身。
2.2 如何解決消息中間件與使用者的強依賴問題
回顧一下解決業務操作和發送消息一致性的方案，會發現我們更多地關注了如何保持和解決一致性的問題，但是忽略了一個問題，那就是消息中間件變成了業務應用的必要依賴。也就是說，如果消息中間件系統（包括使用的消息存儲、業務應用到消息中間件的網絡等）出現問題，就會導致業務操作無法繼續進行，即便當時業務應用和業務操作的資源都是可用的。
我們需要思考如何解決這個問題，思路有如下三種：
•提供消息中間件系統的可靠性，但是沒有辦法保證百分之百可靠。
•對於消息中間件系統中影響業務操作進行的部分，使其可靠性與業務自身的可靠性相同。
•可以提供弱依賴的支持，能夠較好地保證一致性。
第一種方案，提升消息中間件系統的可靠性是必須要做的事情，但是我們無法保證百分之百可靠。
第二種方案，讓消息中間件系統中影響業務操作的部分與業務自身具有同樣的可靠性，其實就是要保證如果業務能操作成功，就需要消息能夠入庫成功。因為如 果消息中間件出問題了，可以接受投遞的延遲，但是需要保證消息入庫，這樣業務操作才可以繼續進行。那么，可行的方式只有一種，如圖6-9所示。
我們把消息中間件所需要的消息表與業務數據表放到同一個業務數據庫中，這 樣，業務應用就可以把業務操作和寫人消息作為一個本地事務來完成，然后再通知消息中間件有消息可以發送，這樣就解決了一致性的問題。從圖6-9中可以看到這一 步是虛線表示的，代表它不是一個必要的操作和依賴。消息中間件會定時去輪詢業務數據庫，找到需要發送的消息，取出內容后進行發送。這個方案對業務系統有如 下三個影響：
•需要用業務自己的數據庫承載消息數據。
•需要讓消息中間件去訪問業務數據庫。
•需要業務操作的對象是一個數據庫，或者說支持事務的存儲，並且這個存儲必須能夠支持消息中間件的需求。
我們在上面的基礎上進行一下變通，如圖6-10所示。這個方案和圖6-9中方案
的區別是，消息中間件不再直接與業務數據庫打交道。消息表還是放在業務數據庫 中，完全由業務數據庫來控制消息的生成、獲取、發送及重試的策略。這樣，消息 中間件就不需要與眾多使用這種消息一致性發送的業務方的數據庫打交道了，不過 比較多的邏輯是從消息中間件的服務端移動到消息中間件的客戶端，並且在業務應 用上執行。消息中間件更多的是管理接收消息的應用，並且當有消息從業務應用發 過來后就只管理投遞，把原來的調度、重投、投遞等邏輯分到了客戶端和服務端 兩邊。
圖6-9和圖6-10中的兩種方式雖然已經解決了大部分問題，但是它們都要求業務操作是支持事務的數據庫操作，具有一定的限制性，這里我們可以再進行一 下變通。
我們考慮把本地磁盤作為一個消息存儲，也就是如果消息中間件不可用，又不願或不能侵入業務自己的數據庫時，可以把本地磁盤作為存儲消息的地方，等待消息中間件回復后，再把消息送到消息中間件中（如圖6-11所示)。所有的投遞、重試等管理，仍然是在消息中間件進行，而本地磁盤的定位只是對業務應用上發送消息 一定成功的一個保證。
這種方式存在的風險是，如果消息中間件不可用，而且寫入本地磁盤的數據也 壞了的話，那么消息就丟失了。這確實是個問題，所以，從業務數據上進行消息補發才是最徹底的容災的手段，因為這樣才能保證只要業務數據在，就一定可以有辦法恢復消息了。
將本地磁盤作為消息存儲的方式有兩種用法，一是作為一致性發送消息的解決方案的容災手段，也就是說該方式平時不工作，出現問題時才切換到該方式上，二 是直接使用該方式來工作，這樣可以控制業務操作本身調用發送消息的接口的處理時間，此外也有機會在業務應用與消息中間件之間做一些批處理的工作。
最后，我們來看一下業務操作與發送消息一致性的方案所帶來的兩個限制。
•需要確定要發送的消息的內容。因為我們在業務操作做之前會把狀態標記為待處理，這要求先能確定消息內容；這里可以有一個變通，即先把主要內容也就是能夠標記該次業務操作特點的信息發過來，然后等業務操作結束后需要更新狀態時再補全內容。不過這還是要求在業務操作之前能夠確定一些索引性質的信息。
•需要實現對業務的檢查。也就是說為了支持反向流程的工作，業務應用必須能夠根據反向流程中發回來的消息內容進行業務操作檢查，確認這個消息所指向的業務操作的狀態是完成、待處理，還是進行中，否則，待處理狀態的消息就無法被處理了。
2.3 消息模型對消息接收的影響
前面講述了消息發送端的內容，我們接下來看一下消息模型。在JMS中，有Queue (點對點）和Topic (發布/訂閱）兩種模型，我們來看看這兩種模型的特點。
2.3.1 JMS Queue模型
圖6-12顯示的是JMS Queue模型，可以看到，應用1和應用2發送消息到JMS 服務器，這些消息根據到達的順序形成一個隊列，應用3和應用4進行消息的消費。 這里需要注意的是，應用3和應用4收到的消息是不同的，也就是說在JMS Queue 的方式下，如果Queue里面的消息被一個應用處理了，那么連接到JMS Queue上的 另一個應用是收不到這個消息的，也就是說所有連接到這個JMS Queue上的應用共同消費了所有的消息。消息從發送端發送出來時不能確定最終會被哪個應用消費， 但是可以明確的是只有一個應用會去消費這條消息，所以JMS Queue模型也被稱為 Peer To Peer (PTP)方式。
2.3.2 JMS Topic模型
圖6-13顯示的是JMS Topic模型。從發送消息的部分和JMS Topic內部的邏輯 來看，JMS Topic和JMS Queue是一樣的，二者最大的差別在於消息接收的部分，在 Topic模型中，接收消息的應用3和應用4是可以獨立收到所有到達Topic的消息的。 JMS Topic模型也被稱為Pub/Sub方式。
2.3.3 JMS中客戶端連接的處理和帶來的限制
在使用JMS時，每個Connection都有一個唯一的Clientld，用於標記連接的唯一性，也就是說剛才對Queue和Topic的介紹中，我們是默認一個接收應用只用了一個連接。現在來看一下多連接的情況，如圖6-14所示。
