有狀態消息隊列在國內氣象通信系統的應用①

胡英楣,王甫棣,譚小華,邢麗平,喬 淼

1(國家氣象信息中心,北京 100081)

2(湖北省氣象信息與技術保障中心,武漢 430074)

3(內蒙古自治區氣象信息中心,呼和浩特 010051)

1 引言

氣象資料是氣象業務和科學研究的基礎[1],地面、高空、衛星、雷達等海量多源觀測數據及產品的實時收集與分發對于氣象資料的共享與使用具有重要意義.氣象通信系統承擔氣象觀測資料、預報預測和服務產品收集與分發,是連接氣象綜合觀測系統、氣象預報預測系統和公共氣象服務系統的橋梁與紐帶,是支撐氣象業務和大氣科學研究的基礎業務平臺[2].

1992 年10 月,氣象衛星綜合應用業務系統(“9210工程”)獲批立項.1998 年,依托9210 工程建設的國內氣象通信業務系統投入業務運行,通信傳輸方式也由衛星通信改為地面寬帶通信.該系統適應當時的計算環境和氣象資料需求進行設計,為保證數據處理的一致性和完整性,采用了單一進程方式依托數據節目表控制來進行氣象資料的接收、處理和分發[3].隨著地球環境觀測、全球氣候觀測系統、全球大氣觀測系統等國際性的觀測網絡的日趨完善以及各類氣象觀測手段的不斷發展,國內參加交換的數據種類、數量、頻次也隨之猛增,尤其是自動站地面觀測、高空探測、多普勒雷達、衛星、閃電定位、GPS、沙塵暴、大氣成分、生態、水文原始觀測數據以及預報產品等新增的氣象數據.9210 國內氣象通信業務系統在擴展性方面的弊端逐漸顯現,原有通信系統在資料的傳輸時效能力上不能滿足業務的需求,國家級和各個省級信息部門自行開發了傳輸系統來滿足新增探測數據的傳輸業務需求.這種單一進程方式的通信業務處理模式是制約系統擴展能力的原因之一.

2 消息隊列技術的應用

2.1 新一代國內氣象通信系統的應用

消息隊列一種線性表,它提供了有保障的消息傳遞、有效的路由、安全性、事務處理支持以及基于優先級的消息傳遞,是用于創建分布式、松散連接的消息通信應用的關鍵技術之一,在氣象通信系統有著廣泛的應用[4].

2006 年,依托中國氣象局氣象寬帶網項目,新一代國內氣象通信系統開展建設[5].該系統為了解決各類新增氣象探測資料的數據傳輸,采用模塊化的設計并內置消息隊列(Message Queue,MQ)方式實現數據收發內部進程通信[6],實現系統內部功能模塊的解耦以支撐多機部署,增強系統適應海量新增資料通信的可伸縮性.

在新一代國內氣象通信系統中,利用消息隊列技術實現通信系統收發任務的并發執行、異步處理以及任務分解/聚合處理等,滿足各類氣象資料不同的處理邏輯以及多用戶多方式傳輸的需求.比如針對氣象資料的預處理以及格式檢查等不需要與主收發進程同步執行的任務啟用一個新的任務消息隊列進行異步處理;將不用用戶接收數據不同的傳輸方式、傳輸路徑切分到不同的任務消息隊列中,也可以將不同的數據針對同一用戶的分發合并到一個任務消息隊列中.

2.2 全國綜合氣象信息共享平臺的應用

依托新一代國內氣象通信系統的建設成果,2009 年,在全國綜合氣象信息共享平臺(China Integrated Meteorological Information Service System,CIMISS)中,建設了數據收集與分發子系統(China Telecommunication System,CTS),它是整個CIMISS 數據處理前端,負責收集來自國內氣象綜合探測系統、互聯網、業務單位、行業部門的各類觀測數據和加工產品,進行規范化的預處理,實現國家級與省級、省級與省級之間數據的互聯互通,保障氣象信息傳輸共享的時效性和可靠性,為預報預測、公共氣象服務等提供有力的數據支撐[7],是國內氣象通信系統的延續.CTS 于2015 年底開始投入業務試運行,2016 年3 月全國業務化.

在CTS 中應用了Active MQ 消息中間件服務器,將消息隊列技術擴展到多機集群間進程間通信.利用可靠異步傳輸機制實現高性能、高可用的消息傳輸中心并對外提供統一標準的公共消息傳輸服務,降低系統的耦合度及管理的復雜性,屏蔽底層異構平臺和跨網絡域業務交互,實現系統不同功能邏輯單元的氣象業務和控制指令消息的消息交互,并推動氣象數據在系統間流轉,實現應用的互連和互操作[8].

基于消息隊列技術的提升,相較于9210 氣象通信系統、新一代國內氣象通信系統,CTS 的布式特性得到增強,是典型的多機多線程模式的分布式系統.在業務運行初期能夠滿足業務的擴展性和實時性等性能要求.但隨著各類氣象資料的統一接入以及服務用戶的不斷增加,系統出現了性能下降甚至出現消息隊列堵塞等問題.

3 有狀態化消息隊列的應用設計

3.1 消息隊列的類型

消息隊列[9-11]按照其是否記錄隊列狀態可劃分為無狀態化消息隊列和有狀態化消息隊列:

(1)無狀態化消息隊列.任何應用進程的操作將發送消息到隊列中,其他系統根據需要訂閱該消息,然后按照需求進行業務邏輯處理.在面向對象(object-oriented)軟件設計開發過程中,設計模式(design pattern)代表了最佳的實踐,是軟件開發人員在軟件開發過程中面臨的一般問題的解決方案.無狀態化消息隊列便是采用了典型的發布訂閱設計模式,一個消息可以有多個消費者.

(2)有狀態化消息隊列.將各類請求按照不同的狀態劃分到不同的隊列,從而使得不同的隊列出現問題后相互不影響;還可以進行優先級區分,一些重要請求可以優先處理等.有狀態化消息隊列按照隊列類型可以延伸包括等待隊列、排重隊列、本地執行隊列、失敗隊列等;按照優先級可以劃分為普通隊列、優先隊列等.有狀態化消息隊列采用了觀察者模式,它屬于行為型模式的一種,通過定義對象間的一種一對多的依賴關系,當一個對象的狀態發生改變時,所有依賴于它的對象都得到通知并被自動更新.將觀察者模式用于氣象信息系統也有不少實踐案例,比如在氣象衛星數據接收與預處理調度機制的設計中應用該模式構建了相應的數據接收與預處理分發作業調度機制[12],在北京全球信息系統中將此模式應用于緩存數據處理[13].

3.2 現有問題的分析

新一代國內氣象通信系統和CTS 第一版的設計采用的無狀態化消息隊列這種模式,在系統資源能力充足的條件下,采用這種方式是最為簡化且保證傳輸效率的模式.

以CTS 為例,圖1 所示為CTS 數據收發的流程:對每一個收發任務(task),數據收集進程(common collection)將準備好處理分發的數據送入處理目錄,并將待處理消息推送到待處理消息隊列中;在不考慮優先級控制條件下,數據處理進程(proc)遵循消息隊列先進先出原則順序處理隊列中的消息;若處理成功則將待分發消息推送到待分發隊列中,并按照待分發目的用戶組織數據目錄;最后數據分發進程(dist)申請分發資源并根據待分發隊列中消息的通知處理用戶的數據分發.

圖1 CTS 優化前數據收發流程圖

作為一個7×24 不間斷運行的業務系統,CTS 也設計了數據收發任務監控并進行相應的容錯處置功能.為了保障收發系統的數據的完整性,通過任務進程反饋的成功與否標志來判定是否重新處理.換句話說,系統并不記錄消息隊列狀態,也不記錄當前實際任務處理情況,通過從消息隊列的訂閱情況來進行錯誤處理.如圖1 所示,當數據處理進程(proc)處理失敗時,會將此消息重新排入隊尾等候下一次處理;對于數據分發進程(dist)若分發失敗時也采取同樣的處理方式.

假設系統的計算資源足夠大,當外部應用消費隊列中消息時,出現異常后發送的回滾消息始終都無法占滿消息隊列,所有的消息最終都會被消費掉.在不計較正常數據的傳輸時效條件下,這種消息的順序處理和異?；貪L策略是保證所有數據傳輸可靠傳輸的可以接受的解決方案.但這種理想狀態下的設計方案沒有考慮復雜的業務情況,特別是在出現氣象寬帶網絡或通信服務器導致的收發系統持續性的異常時.

CTS 每日處理著個數為千萬級別、容量為TB 級別的氣象實時資料,同時數據的時效性是氣象通信系統最重要的評價指標.若按此功能設計,伴隨著CTS接入資料量和分發用戶的逐步增加,CTS 中異常狀態信息會在被處理失敗后不斷追加回消息隊列的末尾,不斷的捕獲異常消息并進行處理會占據大量正常資源,簡單的循環處理造成CTS 處理正常任務能力下降,整體傳輸時效降低.更為關鍵的問題是:如果當這類收發任務出現若干這種持續性異常時,當不斷被處理和回滾的異常消息占滿隊列時,正常的消息將無法再進入隊列,導致隊列陷入死循環狀態且無法自動恢復,CTS一旦癱瘓必須進行人工處理.

3.3 對CTS 的優化改進設計

要想使正常的數據收發不受影響,需要將消息隊列中的異常任務消息及時移除,并且在異常狀態恢復之前,不再向分發隊列中發送異常用戶的消息.考慮引入有狀態化消息隊列的模式改進CTS 數據收發功能:引入一個狀態文件,通過記錄消息隊列中收發狀態信息來進行容錯處置,并且根據狀態文件記錄的不同狀態,對收發系統進行不同的處理邏輯.

如圖2 所示,本文方法在現有收發系統中增加狀態文件,作為一個抽象的“觀察者”,通過擴展抽象類形成收發任務狀態、處理狀態和分發狀態等不同的“觀察者”.每個“觀察者”進一步包含具體的狀態處理操作.比如對于分發狀態,包括對于因為目的用戶分發失敗、申請系統分發資源失敗、配置文件用戶狀態暫停標志、以及數據量超限等;對于處理狀態,可能包括正常狀態、異常狀態以及因異常次數未達閾值的異常狀態等.

圖2 采用有狀態化消息隊列方案的建模

4 應用實現及優化效果

4.1 CTS 優化的應用實現

參照改進方案的建模設計,設計并實現了如圖3所示的CTS 優化流程:對于數據處理未成功的任務,系統將判斷狀態文件的情況,如處于異常狀態且異常次數超過閾值的,將待處理文件寫入磁盤(落盤),更新狀態文件并記錄,以待后續處理;對于數據分發未成功的任務,系統將判斷狀態文件的情況,如處于異常狀態且異常次數超過閾值的,將待分發文件寫入磁盤(落盤),更新狀態文件并記錄,以待后續分發.系統獨立創建數據再處理(reproc)和數據再分發(redist)進程會按照休眠時間策略以及優先級專門處理任務處理狀態為異常的消息,并不斷修改狀態文件中的信息.當狀態文件記錄的任務處理狀態正常時,再處理(reproc)和再分發(redist)進程會重新喚起落盤目錄文件,送入正常的處理流程中.這樣實現的好處在于所有異常的消息不會直接送入正常消息隊列,保證了正常消息隊列的快速處理.同時,對于各類異常狀態的不同處理也使得系統的容錯能力得到增強,提升系統的可靠性.

為提高用戶狀態監控的合理性與準確性,對異常次數進行實時累加和遞減處理,通過閾值范圍提供用戶狀態是否正常的判定條件.

通過這些狀態的排列組合又形成收發任務(task)的多種狀態,在系統實現中通過狀態文件的標志進行區分.

(1)數據處理/再處理(proc/reproc)獲得消息向數據分發(dist)寫入消息文件時,生成.task,表明任務初始化狀態.

(2)數據分發(dist)獲得消息后,準備處理,將任務從.task 修改為.tasking,表明任務的正常狀態.

(3)若用戶狀態文件中分發狀態正常時,而數據分發(dist)申請計算資源失敗時,生成.task.src,表明資源失敗狀態.

(4)用戶狀態文件中分發狀態為異常時,數據處理/再處理(proc/reproc)會向數據分發(dist)目錄下寫入.task.block;數據分發(dist)直接將消息落盤為.task.block,表明任務阻塞狀態.

(5)無論用戶狀態文件中用戶狀態是否正常,一旦數據分發/再分發(dist/redist)發送失敗,都會生成.task.fail,表明任務失敗狀態.

(6)若用戶分發狀態暫停或數據量超限時,數據處理/再處理(proc/reproc)會向數據分發(dist)目錄下寫入.task.susp,表明任務掛起狀態.

4.2 優化的效果

優化后的CTS 的系統穩定性明顯提升,未出現過因異常隊列占滿消息隊列造成CTS 癱瘓的重大故障.同時,本次優化更因為減少了大量不必要的異常消息處理,使得通信傳輸性能明顯提高,數據收集與分發的小時平均時間延遲由優化前的20 s 以上減少至10 s 左右.圖4 選取2016 年5 月(優化前)與2018 年10 月(優化后)CTS 優化前后多日平均各時次傳輸最大時間延遲、最小時間延遲、平均時間延遲進行對比,可以看出,優化后的各項時間延遲較優化前均有明顯的減少.

圖3 CTS 優化后數據收發流程圖

圖4 CTS 優化前后各時次平均傳輸時間延遲對比

5 結束語

通過引入有狀態化消息隊列模式的設計,將異常消息的狀態準確記錄,通過不同的系統進程區分正常消息和異常消息的處理,保證了異常消息不會占據寶貴的正常消息隊列,減少異常消息被反復處理的資源消耗,既提升了通信系統的可靠性,也優化了傳輸時效性.經過CTS 的優化效果也證明了該技術思路的有效性.目前國內氣象通信軟件系統第二版(CTS2.0)正在建設中[14,15],在CTS 的基礎上增加了消息和流兩種通信傳輸模式,提升了實時氣象數據高時效傳輸能力,該優化思路值得在新系統的建設借鑒并運用.