消息隊列在分布式系統數據中的應用與分析

李志剛　劉竹旺

[摘 要]隊列的主要目的是提供路由并保證消息的傳遞，如果發送消息時接收者不可用，消息隊列會保留消息，直到可以成功地傳遞它。消息隊列使應用程序的集成和在不可靠但節省費用的網絡上開發可靠的應用程序變得更加容易。本文主要探討消息隊列在分布式系統數據中的應用與分析。

[關鍵詞]消息隊列 分布式系統 數據

消息隊列是一種進程間通訊機制，本文對單機消息隊列進行了有效的擴展，使其分布在網絡上作為一種基礎中間件平臺為分布式計算提供服務。隨著各種計算機標準的出現和技術的進步，中間件技術作為軟件行業崛起的一個嶄新分支，正在全球范圍內迅猛發展，缺少一個開放的標準是消息中間件不能被大量使用的重要障礙。中間件技術的廣泛運用是在最近10年。消息中間件是中間件范疇中發展最快的中間件類型之一。

一、消息隊列方法的基本設計

消息隊列是操作系統的進程之間用于通信的一種機制，兩個或多個進程間通過訪問共同的消息隊列完成消息的交換。將消息隊列的概念擴展到位于分布式環境下不同站點上應用程序間的通信，由消息隊列機制實現網絡上的可靠消息傳送，其中涉及的消息隊列有消息接收隊列、更新發送表、消息發送管理器和消息接收管理器。

當應用程序在處理數據庫修改或更新事務時，修改更新了本地服務器的數據庫中所保存的數據，同時應用程序將更新信息和更新所執行的SQL（StructuredQueryLanguage）語句以消息的形式寫入到更新發送表，更新發送表的內容和本次本地服務器數據庫更新的內容一致。根據服務器軟件的設定，當服務器發起與其他服務器的數據一致操作時，觸發消息發送管理器將更新發送表的內容封裝成Message消息發送給其他服務器。對本地而言，若無異常，消息將送到目的地址所指示的遠端站點的消息接收隊列中。同時接送端，一個后臺消息接收管理器一直監視著消息接收隊列。當檢測到有消息到達時，開始讀取消息，并根據讀到的消息內容恢復成相應的SQL語句，根據SQL語句對本地數據庫執行所描述的更新事務。如果更新成功，則處理結束，并返回“更新成功”，否則返回“更新失敗”。此時發送端的消息發送管理器一直監視接收隊列，根據收到的消息類型進行相應的操作，若各服務器更新成功，則將該更新發送表內容刪除，若失敗，則周期性地重發該消息。

二、分布式消息隊列操作

分布式消息隊列的寫操作就是向一個已有的消息隊列中發送數據的過程。寫消息之前，用戶進程應獲得其消息隊列描述符msqid的值，然后由msgsnd_d（）系統調用將數據寫入由msqid值指定的消息隊列。具體過程如下：用戶進程調用msgsnd_d（）時，系統調用msgsnd_d（）首先向本地的守護進程發送“寫數據”的請求信息，在請求信息中包含msgsnd_d（）調用的參數msqid、msgflg和msgp，然后等待守護進程返回應答信息，收到應答信息后把結果返回給用戶進程，若是錯誤信息，還要設置相應的錯誤信號errno。守護進程收到“寫數據”的請求后也派生出一個子進程處理。從請求信息中分離出參數msqid、msgflg和要寫入的信息，然后在key值表中查找msqid，若不存在則將錯誤信息返回給用戶進程，若存在則返回其key值表中的key值及所在節點的IP值，然后守護進程根據得到的IP來判斷消息隊列在本地還是在異地。若在本地則調用msgsnd（）向本地消息隊列中寫入數據并將結果返回給用戶進程；若在異地則根據IP向相應的節點發送包括key、msgflg和要寫入信息的“寫數據”請求，在收到應答信息后將結果返回給用戶進程。

三、消息隊列在分布式系統數據中的應用

分布式數據庫系統中的每個網絡節點上，在數據一致性功能模塊下，都由5個功能模塊構成：發送更新消息，接收更新消息，消息發送管理，消息接收管理和更新消息處理。

1.發送更新消息

發送更新消息的功能是向整個網絡所有的服務器發送更新消息數據，由此使其他所有的服務器根據該消息的內容實現與本服務器間的數據一致。發送更新消息通過接收本地服務器所執行的SQL語句中對服務器中數據做出了修改的語句，接收數據修改信息，并將接收的數據修改信息封裝成消息的格式，存儲在更新發送表中，在收到發送命令后，發送給其他服務器，實現各服務器與本服務器的數據同步。

當發送更新消息收到數據庫執行SQL發來的數據修改信息時，先將數據修改信息封裝成消息中的數據信息格式。消息發送命令被觸發后，在消息發送管理器的控制下，將消息以組播的形式發送給其他所有在線服務的業務服務器，實現正常情況下的多服務器間的數據同步。

2.接收更新消息

接收更新消息的功能是實現接收所有數據同步信息（業務層發來的業務數據、其他服務器發來的數據同步消息），并根據收到消息的類型對數據進行相應處理。當收到的數據是業務服務器發送來的業務數據時，對數據加以解析后將解析的數據存儲到相應的緩存區域。當接收到的數據是其他服務器發來的數據同步信息時，對同步數據加以解析，解析消息中的數據信息部分，獲取一次業務的所有數據操作并對本服務器數據庫中的數據執行相應的SQL語句，進行相應的操作，從而實現本服務器與消息發送端服務器之間的數據同步。從而實現了在正常運轉情況下的數據同步操作。

3.消息發送管理

每個服務器在執行對數據庫中的數據進行修改操作的SQL語句后，消息發送管理模塊對更新發布表執行插入操作，添加一條與該條SQL語句相應的消息。服務器觸發進行數據一致過程后，要求與其他服務器進行數據一致，服務器發送數據一致命令給消息發送管理模塊，消息發送管理模塊被觸發后將消息信息發送表的內容取出封裝成用于傳送消息，并根據當前服務器的拓撲信息將消息發送給其他服務器。當消息發送管理器模塊接收到消息收管理模塊發來的重新發送請求時，將消息信息發送表中的數據重新發送給數據同步出錯的服務器。

4.更新消息處理

服務器的接收更新消息模塊收到新消息后，消息接收管理模塊要求接收更新消息模塊將該消息交給更新消息處理模塊。更新消息處理模塊接收到消息后，按照消息編碼過程的反過程解析消息。將消息中的數據信息解析為相應的SQL語句，并根據這一SQL語句對服務器的數據進行更新。如果更新成功則向發送方返回數據更新成功并將本次更新的數據寫入數據交換表，否則返回發送失敗。

參考文獻：

[1]黃姝娟，杜承烈，尤濤.中間件技術實時性能的比較[J].計算機工程，2009，35（11）：32-37.

[2]徐高潮，胡亮，鞠九賓.分布式計算系統[M].北京：高等教育出版社，2004：23-36.

[3]勒敏，劉建輝.分布式數據庫系統數據一致性維護方法[J].科技廣場，2008 （3）：31-33.