基于消息的氣象數據處理系統運維自動化初探

李從英，支亞京,張淑瑩，李 波，楊遠恒

(貴州省氣象信息中心，貴州 貴陽 550002)

0 引言

“數算一體”的天擎系統具備海量數據存儲、全業務貫通、數據應用高效的能力，可直接支撐天氣、氣候、探測、公服、人影等各類氣象應用的云化融入。數據種類覆蓋全國綜合氣象信息共享平臺(China Integrated Meteorological Information Service System,CIMISS)全數據集，數據質量更高，性能大幅提升，具備直接支撐應用云化改造的能力。

天擎系統數據處理在技術和數據種類較CIMISS[1]大幅度提升的同時，保留和繼承了CIMISS的大部分標準規范，是對CIMISS的繼承和發展。數據處理系統(Data Processing Center,DPC)是天擎的重要環節，所有的數據由CTS(Communications Technology System)收集后，都要經過DPC系統的解碼處理，才能及時入庫。因此DPC是保障數據及時處理并入庫的核心系統，DPC運維工作至關重要。

天擎系統建立快速傳輸、實時質控和解碼處理的一體化流程；根據氣象數據的存儲和應用特點、傳輸方式、數據形態設計了9種處理框架。天擎DPC系統主要負責結構化、非結構化以及storm流式等資料數據處理，由18臺服務器支撐，框架多，運維復雜度高，因此，探究DPC系統自動化運維意義重大。

1 RabbitMQ在DPC系統中的應用

數據入庫在整個天擎系統中主要覆蓋消息轉發、消息接收、消息解碼、存儲管理等環節。消息隊列(message queue)即消息中間件，它提供的有保障的消息傳遞、有效的路由、安全性、事務處理支持以及基于優先級的消息傳遞方式[2]，在氣象通信系統中有著廣泛的應用。RabbitMQ由于開源、支持多種語言的客戶端以及適合在分布式系統中消息存儲轉發的場景等特點，非常符合氣象數據傳輸業務需求[3]。為解決觀測資料傳輸頻次和海量數據傳輸問題，選用RabbitMQ消息隊列技術進行數據傳輸[4]。現有數據處理流程為：CTS[5]通過FTP把數據文件推送到數據交換共享目錄，發送通知消息，原始消息發送至RABBIT5671端口，將拆分后的單條消息發送至RABBIT5672端口，后續推送給DPC處理(圖1)。

圖1 數據處理流程

在實際工作中，數據解碼入庫往往會存在一些問題，主要有以下2種情況。一是系統本身特征導致的。由于DPC系統處理程序種類多，且部署在不同的服務器上，很多處理程序因為進程掉線、進程假死等原因，造成消息積壓、不能及時處理，進而導致數據解碼入庫異常。二是人員修改消息轉發配置導致的。人為修改消息轉發配置出錯等原因造成上游收到消息后并沒有轉發到下游的消息隊列中。現有傳統的監控只能在進程是否掉線等粗線條方面進行告警，以上2種異常情況都不能通過現有監控手段抓取到，只能通過人工巡檢檢查時發現，而這時已經嚴重影響到預報和服務等實時業務。鑒于以上現狀，我們開展了DPC數據處理系統自動化運維，以此保障數據的正常及時入庫，為天氣預報和服務業務提供及時的基礎數據保障。

2 研究方法

隊列(Queue)是RabbitMQ的內部對象，用于存儲消息隊列，并將它們轉發給消費者[6]，它是消息的容器，也是消息的終點。消息一直在隊列里等待消費者將其取走。根據消息是積壓還是長時間沒有接收到2種情況，我們需要獲取消息的具體數量。由于RabbitMQ的消息內容可以處理成嚴格的JSON格式數據，然后從中提取關心的字段，以達到監控的目的。本文首次引進了jQuery技術來獲取RabbitMQ消息隊列相關詳細信息，并利用這些信息，可以更好地監控消息隊列。

RabbitMQ消息的持久化[7]設置，使得在消息轉發服務重啟之后，已經存入磁盤的消息也會存在。也就是說，在消息沒有被消費者消費的情況下，消息會一直存在。利用消息的持久化，根據具體消息隊列messages字段，可以獲得消息數量。根據Queue隊列的字段next_seq_id沒有持久化設置，來判定是否存在長期沒有收到消息。

2.1 jQuery技術簡介

jQuery[8]是一款在Linux系統操作中輕量級且靈活的處理JSON數據的工具，它可以接受標準輸入，命令管道或者文件中的JSON數據，經過一系列的過濾器(filters)和表達式的轉換后，形成我們需要的數據結構并標準輸出。這種特性使我們能很容易在Shell腳本中調用它，將數據轉換成理想格式。

要使用jQuery技術來處理消息，首先需要在Linux服務器上安裝jQuery軟件，本項目中使用的是jq-1.5版本。由于是在服務器內部使用，可以直接在github上下載，下載后將安裝文件包上傳服務器解壓縮部署即可。如果服務器可以連接外網，可以在部署的Linux服務器上通過命令yum install jq來安裝。

2.2 自動化運維的設計與實現

消息中Messages字段是記錄消息數量，可以通過以下命令獲取消息數量：

curl --silent -u usernm:passwd "http://ip:15672/api/queues/%2f/queuesname"|jq ".messages"

設置messages大于一定數量時，將對應隊列的名稱記錄到日志中，并自動到DPC處理服務器上找到對應的處理程序，將其重新啟動。

next_seq_id是記錄消息的總數量，消息有持久化設置且不會清零的特征，只要有新的消息來，next_seq_id的值就會增加。利用上述原理，如果next_seq_id在一定時間內沒有變化，將會啟動報警，提示運維人員檢查RabbitMQ沒有收到消息的具體原因。可通過下面的命令行獲取next_seq_id值:

curl --silent -u usernm:passwd "http://ip:15672/api/queues/%2f/queuesname"|jq ".backing_queue_status.next_seq_id"

其中usernm、passwd分別是安裝RabbitMQ時設置的用戶名和密碼；ip是消息服務器部署地址；2f為隊列所在的虛擬主機名；queuesname對應要監控隊列的隊列名。

2.2.1 消息積壓時的處理 消息積壓但處理程序沒有報錯的情況下，說明處理程序雖然顯示在線，但是實際并沒有工作，程序處于假死狀態，現有監控無法檢測。因此程序代碼設定這種情況下，自動到DPC處理服務器上找到對應的處理程序，將其重新啟動。

通過程序判斷，當messages大于10時,監控程序會自動到處理程序所在服務器上重新啟動處理程序,messages的值可以根據實際需求設置。由于自動處理腳本是對幾千個隊列的監測，不能在某個處理程序重啟之后就停止整個腳本的執行，積壓的消息不能迅速處理完，可能會出現多次重新啟動同一個程序的情況，因此在腳本中設置重新啟動程序后休眠20 s,以給處理程序更多的時間處理積壓的消息。當messages值小于10時，等待程序處理消息入庫即可。結構化資料積壓時，腳本會自動到DPC14和DPC15上去重新啟動對應結構化數據解碼程序；非結構化資料積壓時，腳本會自動到DPC04和DPC05上去重新啟動對應非結構化數據解碼程序；半結構化資料積壓時，腳本會自動到DPC06或DPC07上去重新啟動對應半結構化數據解碼程序，需要注意的是DPC06和DPC07服務器上處理的是不同資料的半結構化數據；對于一些省內自有資料，本省將解碼程序部署在DPC18服務器上。在自動化運維處理過程中，消息積壓的隊列名稱、積壓數量以及處理過程都將寫入日志，方便對頻繁出現問題的資料進行原因分析。程序核心代碼protect_messages函數如圖2所示。

圖2 核心代碼

通過以上代碼調用protect_messages，并給其傳參"AGME_PQC_E.0001.0007.R001_001""DPC14" "DPC15""/space/cmadaas/dpc/CMADAAS_DPC_DECODE/exec""E.0001.0007.R001.sh"，再通過定時任務來設定監控頻率，或通過命令守護監控程序一直處于工作狀態。

其中，“AGME_PQC_E.0001.0007.R001_001”表示待監控的隊列名稱，“DPC14,DPC15”表示對應處理程序所在的服務器名稱或處理隊列數據的程序所在服務器名稱，“/space/cmadaas/dpc/CMADAAS_DPC_DECODE/exec”表示處理程序所在服務器上的路徑，“E.0001.0007.R001.sh”表示處理程序名稱。如需要監控多個消息隊列，添加多個調用命令語句即可。

2.2.2 長時間沒有收到消息的處理 由于長時間沒有收到消息的情況，不能通過消息積壓或者程序不在線等方式處理，只有人工巡檢才能發現，因此目前只能通過告警的方式，第一時間通知值班員人工查找原因，以此減少影響時間。本文通過腳本獲取next_seq_id值后，將該機制引入了Zabbix[9-10]的監控[11]告警模塊。將獲取next_seq_id值的腳本文件部署到相應Linux服務器后，通過在Zabbix調用腳本，設置監控項中abschang()函數，當監控的當前時次消息隊列next_seq_id值和上個時次的值之差為零時，就會被抓取告警反饋給運維人員。

3 結論

本文基于消息粒度對數據消息進行監控，在消息積壓的情況下，放棄傳統只告警的監控這一手段，采用自動重啟解碼入庫程序，保障基礎數據及時入庫，再將處理流程寫入日志，供后期原因分析使用。同時對監測長時間沒有收到消息的情況，能及時提醒運維人員，輔助縮短排除故障的時間。實現了DPC處理程序不工作卻不能及時發現并直接處理故障，保障數據及時入庫，減輕了運維人員的工作壓力，提高了工作效率，解決了2種常見故障的監控難點。