RabbitMQ消息隊列技術(shù)在農(nóng)業(yè)氣象測報業(yè)務系統(tǒng)中的應用

成兆金，莊立偉，張媛媛，譚 奎，伍湘秦

(1.日照市氣象局，山東 日照 276826； 2.國家氣象中心，北京 100081； 3.華云信息技術(shù)工程有限公司，北京 100081；4.重慶市豐都縣氣象局，重慶 408200； 5.湖南省氣象局，湖南 長沙 410000)

0 引言

中國氣象局觀測司2009年下發(fā)了農(nóng)業(yè)氣象測報業(yè)務系統(tǒng)(AgMODOS V1.0)，實現(xiàn)了觀測資料的信息化，系統(tǒng)產(chǎn)生的數(shù)據(jù)文件以FTP方式傳輸[1]。隨著自動觀測技術(shù)的迅速發(fā)展，F(xiàn)TP文件傳輸方式難以滿足大容量實時氣象資料的高時效傳輸。

目前，氣象通信傳輸多采用消息隊列技術(shù)，有研究者利用Java語言，采用RabbitMQ消息隊列技術(shù)，設計開發(fā)消息傳輸服務模塊，能支持多種客戶端，臺站通過消息傳輸客戶端將生產(chǎn)的觀測數(shù)據(jù)進行消息封裝后發(fā)布到消息隊列(Message Queue)中，省級消息服務器從隊列中獲取消息后，以秒級的時間推送到國家級消息服務器，進而完成高時效的數(shù)據(jù)傳輸，滿足農(nóng)業(yè)氣象觀測數(shù)據(jù)不斷增加和變化的需求[2-5]。

本文研究的重點是在AgMODOS的基礎上，引入基于Java語言的消息服務機制，研發(fā)基層臺站客戶端的消息服務軟件，改進新XML標準數(shù)據(jù)格式文件傳輸模塊，進行有效的文件傳輸監(jiān)控與管理。

1 資料和方法

消息服務客戶端軟件采用面向?qū)ο蠹夹g(shù)與模塊化結(jié)構(gòu)技術(shù)框架，利用Java語言，采用RabbitMQ消息隊列等技術(shù)，運用XML可擴展標識語言開發(fā)完成[6-9]。

2 結(jié)果與分析

2.1 消息隊列技術(shù)

消息(Message)是消息隊列中的最小單元，實質(zhì)上是一段數(shù)據(jù)，一般作為多個應用程序之間的信息傳輸載體，消息可非常簡單，比如只包含臺站要傳輸?shù)腦ML[10]。消息隊列(Message queue)，也就是消息中間件，可通過其高可靠的傳遞機制進行和平臺無關(guān)的數(shù)據(jù)交換[11]。消息隊列技術(shù)正逐漸成為各行業(yè)通信系統(tǒng)中廣泛使用的核心技術(shù)。由于RabbitMQ采用AMQP傳輸協(xié)議，服務端支持消息隊列的持久性，可以有效保證各個環(huán)節(jié)流轉(zhuǎn)過程的觀測數(shù)據(jù)完整性，同時可提供靈活的路由控制轉(zhuǎn)發(fā)功能，符合氣象數(shù)據(jù)傳輸?shù)臉I(yè)務需求。

2.2 消息傳輸服務客戶端

2.2.1 消息服務模塊架構(gòu)

消息傳輸服務模塊，采用面向?qū)ο笏季S與模塊化技術(shù)，兩層架構(gòu)設計，由交換微代理接口層和交換微代理服務層組成[12]。

交換微代理接口層為消息傳輸服務系統(tǒng)提供管理配置接口，接口可直接嵌入到AgMODOS中，與交換微代理服務進行信息交換。

交換微代理服務層，也就是業(yè)務邏輯層(BLL)，采用RabbitMQ消息中間件框架及common.io文件監(jiān)聽技術(shù)，以消息流發(fā)送的方式實現(xiàn)農(nóng)業(yè)氣象觀測XML標準格式數(shù)據(jù)發(fā)送、數(shù)據(jù)接收、數(shù)據(jù)過濾/封裝、數(shù)據(jù)緩存和數(shù)據(jù)補傳，保證了數(shù)據(jù)內(nèi)容安全完整[13]。

2.2.2 功能設計

消息收發(fā)采用消息中間件RabbitMQ，將數(shù)據(jù)按照規(guī)范的數(shù)據(jù)封裝和文件名規(guī)則輸出。消息服務參數(shù)采用conf.properties文件配置，完成數(shù)據(jù)接收服務器IP、發(fā)送者身份、編報中心、監(jiān)視路徑等消息參數(shù)配置。通過數(shù)據(jù)過濾，對數(shù)據(jù)文件的格式進行檢查及重復判斷，然后進行消息封裝、數(shù)據(jù)發(fā)送、移盤，對發(fā)送成功的數(shù)據(jù)文件轉(zhuǎn)移到Sending及sendback文件夾。發(fā)送失敗，重新調(diào)用回調(diào)接口，進行數(shù)據(jù)補發(fā)，對發(fā)送成功的數(shù)據(jù)設置緩存存儲，交換服務微代理功能結(jié)構(gòu)如圖1所示。

圖1 交換服務微代理功能結(jié)構(gòu)Fig.1 Exchange service micro-agent function structure

在數(shù)據(jù)接入的臺站類型方面，增加區(qū)分主站、備份站的臺站信息。對同一類資料，主測站和備份測站，采用不同的4級編碼來區(qū)分。省級根據(jù)4級編碼來區(qū)分測站類型，對主站、備份測站運行狀況進行監(jiān)視。

在傳輸服務層面，資料的格式檢查增加時效性檢查，對臺站傳輸客戶端的接收目錄中的資料進行時效判斷，對超過規(guī)定時間范圍的資料數(shù)據(jù)，不再進行傳輸發(fā)送。

2.2.3 傳輸流程

臺站通過消息傳輸客戶端，對AgMODOS系統(tǒng)生產(chǎn)的XML標準格式數(shù)據(jù)Z文件進行監(jiān)聽、校驗、封裝，最后以消息的方式傳給省級消息服務端，具體流程如圖2所示。圖中各部分運行過程：AgMODOS軟件生成XML文件后寫入接口目錄，然后由消息傳輸客戶端經(jīng)common.io文件初始化啟動監(jiān)聽，掃描獲得文件后，進行集合唯一性重復判斷，如果文件重復移盤至重復目錄，通過文件名格檢、fileMatch匹配，則將消息屬性封裝后進行消息發(fā)送至省級CTS消息傳輸服務器，省級不落地直接轉(zhuǎn)發(fā)國家級，不通過格檢的文件移盤錯誤目錄。

(1)程序啟動后監(jiān)聽資料入口目錄，對新增實時文件或年度文件實時處理。

common.io文件監(jiān)聽，采用common.io.FileAlterationMonitor文件目錄動態(tài)監(jiān)控技術(shù)，對監(jiān)控目錄進行間隔1 s的實時動態(tài)監(jiān)聽，獲取目標文件以供檢查。

圖2 消息傳輸模塊流程Fig.2 Message transfer module flow chart

FileMonitor m=new FileMonitor(1 000)；//設置監(jiān)控的間隔時間1 s輪詢1次

//指定文件夾，添加監(jiān)聽

String monitorPath=PROP_MAP.get(“monitorPath”)；

m.monitor(monitorPath，new FlumeListener())；

//開啟監(jiān)聽

m.start()；

(2)通過檢驗規(guī)則完成對新增文件的實時檢驗，匹配成功的數(shù)據(jù)進行后續(xù)流程，未通過校驗的文件按照相應規(guī)則進行移盤操作，并記錄對應日志信息。

對于校驗成功的XML標準格式數(shù)據(jù)Z文件，消息傳輸客戶端會把客戶端的配置信息和XML的屬性信息(表1)封裝成RabbitMQ消息屬性和文件內(nèi)容，形成實時消息文件或年度消息文件，封裝的字節(jié)流消息體構(gòu)成一個完整的消息發(fā)送給省級CTS消息服務器的RabbitMQ中，實現(xiàn)臺站與省級、國家級CTS消息服務器的對接。

(3)對匹配成功的數(shù)據(jù)動態(tài)獲取文件內(nèi)容長度，并讀取文件，接著將讀取的XML標準格式數(shù)據(jù)內(nèi)容轉(zhuǎn)碼為base64后存放至byte數(shù)組中封裝為消息體，與消息頭一起進行封裝成RabbitMQ消息發(fā)送給省級CTS的RabbitMQ進行接收。

表1 農(nóng)業(yè)氣象觀測數(shù)據(jù)消息屬性Tab.1 Message properties of agricultural meteorological observation data

(4)如發(fā)送出現(xiàn)異?；蚴r，將文件的處理信息，封裝到map中，同時另開一個線程，開啟定時輪詢功能，以時間間隔為60 s持續(xù)從map中獲取文件信息，嘗試重連并發(fā)送消息，直到消息發(fā)送成功為止。文件發(fā)送成功后將文件信息從該map中移除。對發(fā)送成功的數(shù)據(jù)文件轉(zhuǎn)移到Sending及sendback文件夾，如果發(fā)送失敗，重新調(diào)用回調(diào)接口，進行數(shù)據(jù)補發(fā)。

2.2.4 消息服務傳輸操作流程

消息傳輸是指通過Java消息服務協(xié)議進行網(wǎng)絡內(nèi)或遠程文件傳輸。AgMODOS系統(tǒng)提供簡便、快捷的文件上傳工具。從程序組AgMODOS中選擇AgMOService(或AgMOEditor)程序項，在其主菜單上選擇“XML消息傳輸”，調(diào)用Java程序“MsgTransferClientV1.1.jar”，啟動消息服務，點擊“推送”按鈕，將“未傳”XML文件發(fā)送至省級CTS消息服務器，立即不落地上傳國家級CTS消息服務器，然后查看文件發(fā)生信息，確保消息服務正常，完成臺站發(fā)報任務。如果消息服務未啟動，點擊“檢測消息服務”按鈕，檢查消息服務狀態(tài)，再點擊“啟動消息服務”按鈕，啟動Java消息服務，重新“推送”(圖3)。

圖3 消息服務傳輸操作流程Fig.3 Message service transfer operation flow chart

3 結(jié)論與討論

(1)基于原農(nóng)業(yè)氣象測報業(yè)務系統(tǒng)(AgMODOS1.0)的架構(gòu)技術(shù)，引入Java消息服務傳輸技術(shù)，彌補了FTP傳輸?shù)牟蛔恪?020年12月1日的全國業(yè)務單軌運行以來，表明臺站生成的XML標準文件上傳及時、入庫解譯正確、無丟失，達到了業(yè)務運行要求，對進一步實現(xiàn)氣象數(shù)據(jù)標準化建設具有重要意義。

(2)消息傳輸模塊基于Java語言的jdk1.8.0版本開發(fā)，對臺站計算機的操作系統(tǒng)要求較高，必須Win7及以上版本方可安裝，對環(huán)境變量的配置必須手工操作，且Java的更新升級，必須將計算機操作系統(tǒng)環(huán)境變量“Java_Home”的變量值進行同步更新，更換成更新升級后的版本信息，給臺站業(yè)務人員帶來諸多不便，將來需進一步完善消息服務客戶端軟件，以適應專業(yè)版、企業(yè)版、家庭版、旗艦版等各版本的操作系統(tǒng)，并做到自動配置Java的系統(tǒng)環(huán)境變量。

(4)消息傳輸系統(tǒng)是中國氣象局推行業(yè)務標準化建設新開發(fā)的業(yè)務系統(tǒng)，是一個新概念，要求臺站業(yè)務人員認知消息服務的概念和有關(guān)知識。業(yè)務準入運行以來：系統(tǒng)運行穩(wěn)定，Z文件數(shù)據(jù)99%在1 s內(nèi)完成從臺站到國家局的傳輸，達到了業(yè)務運行要求。