消息隊列技術在氣象軟件系統中的應用

■ 王恩文

國內氣象通信軟件系統第二版（CTS2）中的消息傳輸模塊已經在全國2000多個氣象站進行業務運行。目前全國所有的國家級地面自動站BUFR數據均通過該系統進入國家級、省級的氣象數據環境，并且為省級的MDOS系統提供了數據服務支撐。從試點運行到現在的業務運行，消息傳輸運行穩健，90%的數據從臺站到國家的傳輸時間在2 s以內，99.99%的數據在10 s以內完成傳輸。

國內氣象通信系統是承擔國內氣象資料和產品收集以及國內外氣象資料和產品的國內分發的業務系統。隨著現代氣象業務快速發展，氣象數據的種類、數量日益增加，關鍵業務資料觀測、更新頻次達到分鐘級，傳輸時效要求達到秒級。現有國內氣象通信系統，包括目前業務運行的新一代國內通信系統、國內氣象通信軟件系統第一版（CMA domestic telecommunication system（version 1），CTS1），僅支持國家級和省級間基于文件的數據傳輸，難以滿足未來的統一省以下通信傳輸技術體制，難以支持多種類、高頻次、大容量實時氣象資料的高時效傳輸。

國內氣象通信軟件系統第二版（CMA domestic telecommunication system（version 2），CTS2），在現有系統基礎上新增消息傳輸、流傳輸模式，并以Redis消息隊列（Redis message queue，RMQ）為核心的交換控制系統替換現有的調度系統，大幅度提升了氣象數據實時傳輸效率。

1 消息隊列技術

1.1 消息隊列基本概念

消息（Message）是消息隊列中最小的概念，本質上是一段數據，它能被一個或者多個應用程序所理解，是應用程序之間傳遞的信息載體，消息可以非常簡單，比如只包含文本字符串、XML、JSON等，也可以非常復雜，比如圖片、BUFR（Binary universal form for representation of meteorological data）編碼的二進制文件。

消息隊列（Message queue，MQ），也可以稱為消息隊列中間件或消息中間件，是指利用高效可靠的消息傳輸機制進行與平臺無關的數據交流，并給予數據通信進行分部署系統的集成。通過提供消息傳遞和消息排隊模型，它可以在分布式環境下擴展進程間通信。

消息隊列一般有兩種傳遞模式：點對點（P2P，Point-to-Point）模式和發布/訂閱（Pub/Sub）模式。點對點模式是基于隊列的，消息生產者發布消息到對壘，消息消費者從隊列中接收消息，隊列的存在使得消息的異步傳輸成為可能。發布/訂閱模式定義了如何向一個內容節點發布和訂閱消息，這個內容節點稱為主題（Topic），主題可以認為是消息傳遞的中介，消息發布者將消息發布到某個主題， 而消息訂閱者則從主題中訂閱消息。主題使得消息的訂閱者與消息的發布者保持獨立，不需要進行接觸即可保證消息的傳遞，發布/訂閱模式在消息的一對多廣播時采用。

1.2 采用RabbitMQ的原因

目前市場上的消息中間件有很多，比較主流的消息隊列有RabbitMQ、ActiveMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ。

在系統設計初期進行了大量的業務環節消息中間件選型測試，從以下方面進行選型和對比：

1)傳輸效率：在所選的6類消息MQ中傳輸效率ZeroMQ最高，RabbitMQ次之。

2）持久化消息：ZeroMQ不支持，ActiveMQ和RabbitMQ都支持。持久化消息主要是指機器在不可抗力因素等情況下掛掉后，消息不會丟失的機制。

3）綜合技術實現：可靠性、靈活的路由、集群、事務、高可用的隊列、消息排序、問題追蹤、可視化管理工具、插件系統等。RabbitMQ/ Kafka 最好，ActiveMQ次之，ZeroMQ最差。當然ZeroMQ也可以做到，不過必須手動寫代碼實現，代碼量不小。

4）高并發：RabbitMQ最高，因為它的實現語言是天生具備高并發、高可用的erlang語言。

經過綜合考慮，RabbitMQ最符合業務需求。

1.3 消息隊列的作用

消息隊列憑借其獨到的特性，在不同的應用場景下可以展現不同的作用：

解耦：消息隊列在處理過程中間插入了一個隱含的基于數據的接口層，這允許程序獨立地擴展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束即可。CTS2中消息通信解耦了數據傳輸線路中各環節的生成、入庫等應用的強關聯，各應用只須做好自己的工作，無須擔心向下游的程序是否正常，簡化了系統架構。

冗余（持久化）：處理數據有時進程失敗。除非數據持續存在，否則它將永遠丟失。消息隊列通過持久化數據，直到完全處理來緩解這一點。許多消息隊列使用的put-get-delete范例需要一個進程來明確指出在郵件從隊列中刪除之前已經完成了處理消息，確保數據保持安全，直到完成。應用場景：CTS2中正是使用這一特性來保證數據的完整性。

可擴展性：消息隊列解耦了應用的處理過程，所以可以輕松地擴展。將消息添加到隊列，只須添加另一個進程。不需要更改代碼，不需要調整任何配置。應用場景：CTS2中利用這一特性，可使國家級、省級各個平臺根據直接的數據規模配置適量的消息隊列服務器（Broker）即可。

彈性（流控）：消息隊列能夠使關鍵主鍵支撐突發訪問壓力，不會因為突發的超負荷請求而完全崩潰。如果正在處理來自隊列的消息的進程失敗，那么仍然可以將消息添加到隊列中以在系統恢復時進行處理。應用場景：在CTS2中這一特性可以使當數據源端異常時（比如傳輸客戶端因配置錯誤突然導入多年歷史數據），整個傳輸系統可以保證穩定運行。

緩沖：在任何重要的系統中，都需要不同處理時間的組件。消息隊列通過提供緩沖層來幫助這些任務以最高效率運行，該緩沖區有助于控制和優化數據流經系統的速度。應用場景：CTS2系統中數據源程序、數據處理程序、數據入庫程序無須保證同時啟動，任何環節的應用再啟動后只要有任務或數據即可進入工作狀態。

異步通信：消息隊列啟用異步處理，允許將消息放在隊列上，而不是立即處理。可以隨意添加多個消息，然后隨意處理。這樣可使CTS2在下游系統未啟動的情況下，仍然正常收發。

2 RMQ在CTS2中的應用

在CTS2中，RMQ消息傳輸承擔著全局任務管理、數據收集任務管理、數據分發任務管理、消息數據傳輸的總要任務（圖1），是國家級地面自動站BUFR格式消息數據的唯一傳輸途徑，是全局任務管理的核心。

圖1 消息隊列在CTS2系統架構中的應用

2.1 消息傳輸架構設計

國內氣象通信系統基于RabbitMQ開發傳輸客戶端和消息服務端，其中傳輸客戶端部署到臺站，消息服務端部署到省級和國家級，傳輸流程如圖2所示（以地面自動站BUFR格式數據為例）。

2.1.1 消息數據傳輸基本架構

國家級地面自動站BUFR格式消息數據傳輸流程：

1）傳輸客戶端與ISOS實現消息API接口和文件目錄接口，將獲取的消息數據上傳至省級消息服務端。

2）省級MDOS（Meterological Data Operational System）調用消息接口從消息服務端獲取消息數據，并將質控后的消息數據送至消息服務端，消息數據入省級數據環境，消息收發處理進程從省級數據緩存獲取消息數據，導出成文件，同時在省級通過配置shovel，將消息數據shovel到國家級RabbitMQ隊列。

3）消息數據入國家級數據環境。

消息數據傳輸主要包括臺站數據上行和國家級控制指令下行。

全國基礎氣象臺站的各種觀測資料通過傳輸客戶端將數據傳輸至本省省級通信系統，省級中心將省（區、市）的全部觀測資料、加工產品以及其他有關信息上傳至國家級通信系統。

圖2 數據傳輸流程圖

國家級業務系統由國家級傳輸客戶端發送下行指令，通過國家級、省級通信系統，由臺站級傳輸客戶端提供給臺站綜合觀測業務系統。臺站業務系統響應指令后，通過臺站傳輸客戶端發送上行響應，國家級通信系統通過國家級傳輸客戶端向該國家級業務系統進行反饋，同時抄送國家級和省級監視系統（客戶端）。

圖3 物理部署圖

圖4 數據傳輸時效分析圖

省級業務系統由省級傳輸客戶端發送下行指令，通過省級通信系統，由臺站級傳輸客戶端提供給臺站綜合觀測業務系統。臺站業務系統響應指令后，通過臺站傳輸客戶端發送上行響應，省級通信系統通過省級傳輸客戶端向該省級業務系統進行反饋，同時抄送國家級和省級監視系統（客戶端）。

2.1.2 高時效設計

為了滿足國家級地面自動站BUFR數據的秒級傳輸及以后系統消息傳輸數據量可擴展的需求，設計了RMQ多機多節點的集群模式（圖3）。

圖4為雙機多節點的數據傳輸時效性測試圖。通過大量的模擬和業務測試繪制出時效性曲線。可以看出，在所實施的測試條件下，無論每臺機器部署幾個節點，都能很好地滿足傳輸時效性小于2 s的要求。若數據的傳輸時效性要求小于1 s，從測試曲線看出，每臺機器部署8個節點，同時部署2臺服務器即可滿足地面自動站BUFR數據的秒級傳輸要求，如需要對業務進行擴展，只需成對增加服務即可。

2.1.3 安全可靠性設計

RMQ多節點多集群模式除了可以大幅度提升傳輸效率，還可以允許發布者與生產者在單節點崩潰的情況下繼續運行，當集群內失去一個RMQ節點時，客戶端能重新連接到集群的任何其他節點并繼續生產或消費。

另外，在傳輸客戶端進行生產（發送數據）或消費（接收數據）時，分別采用了發布確認模式和接收確認模式。發布確認模式可以保障消息到達消息服務器，當指定時間內無法到達消息服務器會返回發送超時代碼，客戶端可以嘗試進行重新發送。接收確認模式在消費者接收消息后消息體仍然存在于消息隊列中并且被標識為“Unacked”，直到消費者處理完消息后發送“ack”指令，消息體才會從隊列中刪除，如果消費者異常崩潰消息會重新恢復正常狀態供其他消費者消費，從而保證消息不丟失。

圖5 交換控制流程圖

2.2 交換控制

交換控制配置項包括收集任務管理、分發任務管理、編輯任務管理和元數據同步任務管理四個部件（圖5）。各任務管理部件實時讀取內存中的策略配置，解析其中的時間策略和操作環節，根據時間策略觸發生成作業任務，并按照任務優先級將作業送入到相應消息隊列中。相關配置項從對應消息隊列中獲取作業任務，解析任務策略并調度任務執行，當前環節操作完成后，將按照后續流程將數據對象向下游傳遞。

3 結語

CTS2中的消息傳輸模塊已經在全國31個省（區、市）的2000多個國家站進行業務運行，目前全國所有的國家級地面自動站BUFR數據均通過該系統進入國省的氣象數據環境，并且為省級的MDOS系統提供了數據服務支撐。從試點運行到現在的業務運行，消息傳輸運行穩健，90%的數據從臺站到國家的傳輸時間在2 s以內，99.99%的數據在10 s以內完成傳輸。

CTS2中的交換控制模塊也在國家級、貴州省、湖北省開展了試點運行，承擔著3個試驗點的文件傳輸、消息傳輸、流傳輸的調度工作，到目前為止運行穩定，無重大故障發生。

深入閱讀

陳玉華, 翟穎佳, 周紅, 等, 2013. 基于氣象通信系統切換信息傳輸流程梳理與優化. 農業網絡信息, (7): 22-24.

林潤生, 孫周軍, 譚小華, 等, 2011. 新一代國內氣象通信系統設計與實現. 氣象, 37(3): 356-362.

劉金霞, 王慧瑜, 趙威, 等, 2012. 省級新一代氣象通信系統傳輸流程的設計與實現. 氣象與環境學報, 28(6): 76-80.

向筱銘, 徐曉莉, 宋智, 等, 2017. 基于CTS的臺站上行氣象數據傳輸監控平臺的設計與實現. 氣象科技, 45(4): 647-652.

張來恩, 王鵬, 韓鑫強, 2018. CTS2.0消息封裝及交換控制策略設計及實踐. 氣象科技進展, 8(1): 271-273.