有狀態(tài)消息隊(duì)列在國(guó)內(nèi)氣象通信系統(tǒng)的應(yīng)用①

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

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

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

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

1 引言

氣象資料是氣象業(yè)務(wù)和科學(xué)研究的基礎(chǔ)[1],地面、高空、衛(wèi)星、雷達(dá)等海量多源觀測(cè)數(shù)據(jù)及產(chǎn)品的實(shí)時(shí)收集與分發(fā)對(duì)于氣象資料的共享與使用具有重要意義.氣象通信系統(tǒng)承擔(dān)氣象觀測(cè)資料、預(yù)報(bào)預(yù)測(cè)和服務(wù)產(chǎn)品收集與分發(fā),是連接氣象綜合觀測(cè)系統(tǒng)、氣象預(yù)報(bào)預(yù)測(cè)系統(tǒng)和公共氣象服務(wù)系統(tǒng)的橋梁與紐帶,是支撐氣象業(yè)務(wù)和大氣科學(xué)研究的基礎(chǔ)業(yè)務(wù)平臺(tái)[2].

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

2 消息隊(duì)列技術(shù)的應(yīng)用

2.1 新一代國(guó)內(nèi)氣象通信系統(tǒng)的應(yīng)用

消息隊(duì)列一種線性表,它提供了有保障的消息傳遞、有效的路由、安全性、事務(wù)處理支持以及基于優(yōu)先級(jí)的消息傳遞,是用于創(chuàng)建分布式、松散連接的消息通信應(yīng)用的關(guān)鍵技術(shù)之一,在氣象通信系統(tǒng)有著廣泛的應(yīng)用[4].

2006 年,依托中國(guó)氣象局氣象寬帶網(wǎng)項(xiàng)目,新一代國(guó)內(nèi)氣象通信系統(tǒng)開展建設(shè)[5].該系統(tǒng)為了解決各類新增氣象探測(cè)資料的數(shù)據(jù)傳輸,采用模塊化的設(shè)計(jì)并內(nèi)置消息隊(duì)列(Message Queue,MQ)方式實(shí)現(xiàn)數(shù)據(jù)收發(fā)內(nèi)部進(jìn)程通信[6],實(shí)現(xiàn)系統(tǒng)內(nèi)部功能模塊的解耦以支撐多機(jī)部署,增強(qiáng)系統(tǒng)適應(yīng)海量新增資料通信的可伸縮性.

在新一代國(guó)內(nèi)氣象通信系統(tǒng)中,利用消息隊(duì)列技術(shù)實(shí)現(xiàn)通信系統(tǒng)收發(fā)任務(wù)的并發(fā)執(zhí)行、異步處理以及任務(wù)分解/聚合處理等,滿足各類氣象資料不同的處理邏輯以及多用戶多方式傳輸?shù)男枨?比如針對(duì)氣象資料的預(yù)處理以及格式檢查等不需要與主收發(fā)進(jìn)程同步執(zhí)行的任務(wù)啟用一個(gè)新的任務(wù)消息隊(duì)列進(jìn)行異步處理;將不用用戶接收數(shù)據(jù)不同的傳輸方式、傳輸路徑切分到不同的任務(wù)消息隊(duì)列中,也可以將不同的數(shù)據(jù)針對(duì)同一用戶的分發(fā)合并到一個(gè)任務(wù)消息隊(duì)列中.

2.2 全國(guó)綜合氣象信息共享平臺(tái)的應(yīng)用

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

在CTS 中應(yīng)用了Active MQ 消息中間件服務(wù)器,將消息隊(duì)列技術(shù)擴(kuò)展到多機(jī)集群間進(jìn)程間通信.利用可靠異步傳輸機(jī)制實(shí)現(xiàn)高性能、高可用的消息傳輸中心并對(duì)外提供統(tǒng)一標(biāo)準(zhǔn)的公共消息傳輸服務(wù),降低系統(tǒng)的耦合度及管理的復(fù)雜性,屏蔽底層異構(gòu)平臺(tái)和跨網(wǎng)絡(luò)域業(yè)務(wù)交互,實(shí)現(xiàn)系統(tǒng)不同功能邏輯單元的氣象業(yè)務(wù)和控制指令消息的消息交互,并推動(dòng)氣象數(shù)據(jù)在系統(tǒng)間流轉(zhuǎn),實(shí)現(xiàn)應(yīng)用的互連和互操作[8].

基于消息隊(duì)列技術(shù)的提升,相較于9210 氣象通信系統(tǒng)、新一代國(guó)內(nèi)氣象通信系統(tǒng),CTS 的布式特性得到增強(qiáng),是典型的多機(jī)多線程模式的分布式系統(tǒng).在業(yè)務(wù)運(yùn)行初期能夠滿足業(yè)務(wù)的擴(kuò)展性和實(shí)時(shí)性等性能要求.但隨著各類氣象資料的統(tǒng)一接入以及服務(wù)用戶的不斷增加,系統(tǒng)出現(xiàn)了性能下降甚至出現(xiàn)消息隊(duì)列堵塞等問題.

3 有狀態(tài)化消息隊(duì)列的應(yīng)用設(shè)計(jì)

3.1 消息隊(duì)列的類型

消息隊(duì)列[9-11]按照其是否記錄隊(duì)列狀態(tài)可劃分為無狀態(tài)化消息隊(duì)列和有狀態(tài)化消息隊(duì)列:

(1)無狀態(tài)化消息隊(duì)列.任何應(yīng)用進(jìn)程的操作將發(fā)送消息到隊(duì)列中,其他系統(tǒng)根據(jù)需要訂閱該消息,然后按照需求進(jìn)行業(yè)務(wù)邏輯處理.在面向?qū)ο?object-oriented)軟件設(shè)計(jì)開發(fā)過程中,設(shè)計(jì)模式(design pattern)代表了最佳的實(shí)踐,是軟件開發(fā)人員在軟件開發(fā)過程中面臨的一般問題的解決方案.無狀態(tài)化消息隊(duì)列便是采用了典型的發(fā)布訂閱設(shè)計(jì)模式,一個(gè)消息可以有多個(gè)消費(fèi)者.

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

3.2 現(xiàn)有問題的分析

新一代國(guó)內(nèi)氣象通信系統(tǒng)和CTS 第一版的設(shè)計(jì)采用的無狀態(tài)化消息隊(duì)列這種模式,在系統(tǒng)資源能力充足的條件下,采用這種方式是最為簡(jiǎn)化且保證傳輸效率的模式.

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

圖1 CTS 優(yōu)化前數(shù)據(jù)收發(fā)流程圖

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

假設(shè)系統(tǒng)的計(jì)算資源足夠大,當(dāng)外部應(yīng)用消費(fèi)隊(duì)列中消息時(shí),出現(xiàn)異常后發(fā)送的回滾消息始終都無法占滿消息隊(duì)列,所有的消息最終都會(huì)被消費(fèi)掉.在不計(jì)較正常數(shù)據(jù)的傳輸時(shí)效條件下,這種消息的順序處理和異?；貪L策略是保證所有數(shù)據(jù)傳輸可靠傳輸?shù)目梢越邮艿慕鉀Q方案.但這種理想狀態(tài)下的設(shè)計(jì)方案沒有考慮復(fù)雜的業(yè)務(wù)情況,特別是在出現(xiàn)氣象寬帶網(wǎng)絡(luò)或通信服務(wù)器導(dǎo)致的收發(fā)系統(tǒng)持續(xù)性的異常時(shí).

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

3.3 對(duì)CTS 的優(yōu)化改進(jìn)設(shè)計(jì)

要想使正常的數(shù)據(jù)收發(fā)不受影響,需要將消息隊(duì)列中的異常任務(wù)消息及時(shí)移除,并且在異常狀態(tài)恢復(fù)之前,不再向分發(fā)隊(duì)列中發(fā)送異常用戶的消息.考慮引入有狀態(tài)化消息隊(duì)列的模式改進(jìn)CTS 數(shù)據(jù)收發(fā)功能:引入一個(gè)狀態(tài)文件,通過記錄消息隊(duì)列中收發(fā)狀態(tài)信息來進(jìn)行容錯(cuò)處置,并且根據(jù)狀態(tài)文件記錄的不同狀態(tài),對(duì)收發(fā)系統(tǒng)進(jìn)行不同的處理邏輯.

如圖2 所示,本文方法在現(xiàn)有收發(fā)系統(tǒng)中增加狀態(tài)文件,作為一個(gè)抽象的“觀察者”,通過擴(kuò)展抽象類形成收發(fā)任務(wù)狀態(tài)、處理狀態(tài)和分發(fā)狀態(tài)等不同的“觀察者”.每個(gè)“觀察者”進(jìn)一步包含具體的狀態(tài)處理操作.比如對(duì)于分發(fā)狀態(tài),包括對(duì)于因?yàn)槟康挠脩舴职l(fā)失敗、申請(qǐng)系統(tǒng)分發(fā)資源失敗、配置文件用戶狀態(tài)暫停標(biāo)志、以及數(shù)據(jù)量超限等;對(duì)于處理狀態(tài),可能包括正常狀態(tài)、異常狀態(tài)以及因異常次數(shù)未達(dá)閾值的異常狀態(tài)等.

圖2 采用有狀態(tài)化消息隊(duì)列方案的建模

4 應(yīng)用實(shí)現(xiàn)及優(yōu)化效果

4.1 CTS 優(yōu)化的應(yīng)用實(shí)現(xiàn)

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

為提高用戶狀態(tài)監(jiān)控的合理性與準(zhǔn)確性,對(duì)異常次數(shù)進(jìn)行實(shí)時(shí)累加和遞減處理,通過閾值范圍提供用戶狀態(tài)是否正常的判定條件.

通過這些狀態(tài)的排列組合又形成收發(fā)任務(wù)(task)的多種狀態(tài),在系統(tǒng)實(shí)現(xiàn)中通過狀態(tài)文件的標(biāo)志進(jìn)行區(qū)分.

(1)數(shù)據(jù)處理/再處理(proc/reproc)獲得消息向數(shù)據(jù)分發(fā)(dist)寫入消息文件時(shí),生成.task,表明任務(wù)初始化狀態(tài).

(2)數(shù)據(jù)分發(fā)(dist)獲得消息后,準(zhǔn)備處理,將任務(wù)從.task 修改為.tasking,表明任務(wù)的正常狀態(tài).

(3)若用戶狀態(tài)文件中分發(fā)狀態(tài)正常時(shí),而數(shù)據(jù)分發(fā)(dist)申請(qǐng)計(jì)算資源失敗時(shí),生成.task.src,表明資源失敗狀態(tài).

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

(5)無論用戶狀態(tài)文件中用戶狀態(tài)是否正常,一旦數(shù)據(jù)分發(fā)/再分發(fā)(dist/redist)發(fā)送失敗,都會(huì)生成.task.fail,表明任務(wù)失敗狀態(tài).

(6)若用戶分發(fā)狀態(tài)暫停或數(shù)據(jù)量超限時(shí),數(shù)據(jù)處理/再處理(proc/reproc)會(huì)向數(shù)據(jù)分發(fā)(dist)目錄下寫入.task.susp,表明任務(wù)掛起狀態(tài).

4.2 優(yōu)化的效果

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

圖3 CTS 優(yōu)化后數(shù)據(jù)收發(fā)流程圖

圖4 CTS 優(yōu)化前后各時(shí)次平均傳輸時(shí)間延遲對(duì)比

5 結(jié)束語(yǔ)

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