摘? 要：本文主要論述了如何利用消息中間件ActiveMQ，構(gòu)建一個通用的數(shù)據(jù)傳輸框架，以解決同步傳輸效率低下、數(shù)據(jù)傳輸不及時、數(shù)據(jù)傳輸不可靠、難以按優(yōu)先級傳輸數(shù)據(jù)等難題，實現(xiàn)一個可跨平臺、適用于各個異構(gòu)分布式系統(tǒng)、屏蔽底層細節(jié)的數(shù)據(jù)傳輸框架的搭建。該框架在應用過程中提高了業(yè)務處理效率、降低了開發(fā)成本。

關(guān)鍵詞：中間件;ActiveMQ;通用數(shù)據(jù)傳輸框架;民航氣象數(shù)據(jù)傳輸

中圖分類號：TP311.52? ? ? 文獻標識碼：A 文章編號：2096-4706（2019）10-0026-03

Abstract：This paper mainly discusses how to use a Message-oriented middleware ActiveMQ to build a universal data transmission framework. The framework can solve the problems of low efficiency of synchronous transmission，untimely data transmission，unreliable data transmission，and difficulty in transmitting data according to priority. A cross-platform，heterogeneous distributed system，shielding the underlying details of the data transmission framework is implemented. The framework improves business processing efficiency and reduces development costs in the application process.

Keywords：message-oriented middleware;ActiveMQ;data transmission framework;civil aviation meteorological data transmission

0? 引? 言

民用航空氣象服務與民用航空的安全和經(jīng)濟效益密切相關(guān)，隨著民用航空的高速發(fā)展，對氣象服務的要求越來越高。而氣象數(shù)據(jù)種類多樣、報文格式復雜、數(shù)據(jù)量大以及高時效性、準確度要求高[1]等因素給氣象服務帶來了更大的挑戰(zhàn)。而現(xiàn)如今，越來越多的部門對民航氣象數(shù)據(jù)有著迫切的需求。

在當前的業(yè)務條件下，民航氣象數(shù)據(jù)傳輸給用戶主要通過FTP的方式，F(xiàn)TP傳輸對于點對點數(shù)據(jù)傳輸有著方便易用等優(yōu)點，卻存在傳輸及時性不強、可靠性不足、無法配置優(yōu)先級等缺點。

民航業(yè)務運行急需一個傳輸框架，既可以解決各部門中的分布式系統(tǒng)的異構(gòu)問題，屏蔽操作系統(tǒng)的底層細節(jié)，又能保證各類數(shù)據(jù)異步傳輸、即到即轉(zhuǎn)、滿足數(shù)據(jù)時效精確到秒的要求，同時做到100%不丟數(shù)據(jù)，且能設置不同的優(yōu)先級進行數(shù)據(jù)傳輸接收。

基于現(xiàn)實迫切的需求，構(gòu)建了一個基于消息中間件的通用傳輸框架，該框架可異步傳輸數(shù)據(jù)，可跨平臺使用，屏蔽底層細節(jié)，同時做到數(shù)據(jù)可靠傳輸，適用于所有需要傳輸數(shù)據(jù)的場合。該框架選用了基于JMS（Java領域的MOM規(guī)范）實現(xiàn)的中間件——ActiveMQ。

1? ActiveMQ介紹

ActiveMQ是目前最為流行和強大的開源消息中間件（MOM），由Apache軟件基金會（ASF）出品。ActiveMQ是用于系統(tǒng)之間遠程通信的消息代理器，它遵循JMS 1.1規(guī)范和J2EE 1.4規(guī)范中的JMS服務。ActiveMQ是用Java語言實現(xiàn)的，它也為多種語言提供了API，如Java、C、C++、C#、Ruby、Perl、Python和PHP等[2]，還支持OpenWire、Stomp、REST、WS-Notification、XMPP和AMQP等多種應用連接協(xié)議。

在JMS規(guī)范中，消息類型有六種，分別是Message、TextMessage、MapMessage、BytesMessage、Stream-Message、ObjectMessage。在ActiveMQ中，還定義了新的消息類型——BlobMessage[3]，可應用于發(fā)送大文件。

關(guān)于數(shù)據(jù)傳輸模型，ActiveMQ支持兩種：點對點模型（Point To Point）和發(fā)布訂閱模式（Publisher To Sub-scriber）。

第一種模型是點對點模型（Point To Point），這個模型的特點是一對一，在這個模型中，生產(chǎn)者首先封裝數(shù)據(jù)生成消息，然后將消息發(fā)送至Queue，即隊列中。Active MQ服務將根據(jù)傳送策略，將該隊列中的消息傳輸?shù)较虼岁犃凶缘哪骋粋€消費者，一個消息只能傳送一次、只能由一個消費者使用。但是，同一個隊列允許一個至多個生產(chǎn)者向其發(fā)送消息。

第二種模型是發(fā)布訂閱模型（Publisher To Sub-scriber），該模型的特點是一對多，即消息只需生產(chǎn)一次，就可提供個多個消費者即用戶使用。在這個模型中，生產(chǎn)者封裝消息發(fā)送到相應的Topic（目標主題），ActiveMQ會將消息發(fā)送給訂閱了該主題的消費者。可以向Topic發(fā)送消息的生產(chǎn)者也是可以多個的。

發(fā)布訂閱模型還支持持久訂閱，如果消費者向Active-MQ的某個Topic提交持久訂閱要求，即向該主題注冊，注冊后，ActiveMQ將保證該主題的消息能可靠地傳輸給該消費者。當消費者處于掉線狀態(tài)，無法及時接收到消息，ActiveMQ將為其保存該消息，直至消息到期，或者消費者上線，正常接收到消息。未向Topic注冊的消費者掉線時，ActiveMQ不會為其保存消息，即使消費者再次成功連接，之前的消息也將丟失。

ActiveMQ的主要特性有：

（1）遵循JMS1.1規(guī)范。ActiveMQ能夠提供同步和異步的消息投遞、一次且僅一次的消息投遞以及消息訂閱者的消息持久訂閱等功能。

（2）支持多種連接協(xié)議和連接方式。ActiveMQ同時支持不同消息代理器之間的連接和消息代理器與客戶端之間的連接。

（3）支持消息的持久存儲。ActiveMQ能將消息持久化地存儲到本地數(shù)據(jù)庫，以保證消息的有效投遞。

（4）具有安全認證和授權(quán)功能。ActiveMQ可以通過本地配置文件或者插件來實現(xiàn)客戶端的安全認證和授權(quán)，以保證消息只能被授權(quán)用戶生產(chǎn)和消費。

（5）其他高級特性。ActiveMQ還支持許多高級特性，包括消息代理器（broker）的集群、獨享消費者、消息組和大文件消息等等。

2? 基于ActiveMQ的通用數(shù)據(jù)傳輸框架設計與實現(xiàn)

2.1? 架構(gòu)設計

本框架分為業(yè)務邏輯層、控制層、傳輸層、數(shù)據(jù)交換層[4]、分布式系統(tǒng)等幾個部分。為了實現(xiàn)全雙工通信，保證數(shù)據(jù)即到即處理，本框架的實現(xiàn)中對數(shù)據(jù)的處理選用了異步的模式，在控制模塊采用多線程的方式，在傳輸模塊選用了Pub/Sub（發(fā)布者/訂閱者）模型。

如圖1所示，控制模塊負責與業(yè)務邏輯對接，提供數(shù)據(jù)接口，相應業(yè)務按照指定格式，通過調(diào)用數(shù)據(jù)接口將要傳輸?shù)臄?shù)據(jù)發(fā)送給控制模塊，控制模塊根據(jù)配置文件，針對不同的業(yè)務邏輯進行數(shù)據(jù)的封裝，由傳輸模塊作為生產(chǎn)者將消息發(fā)送到指定的主題中。

而作為消息的消費者，傳輸模塊持續(xù)監(jiān)聽指定主題，當主題中有最新的消息時，自動觸發(fā)控制模塊中的處理線程，將數(shù)據(jù)解封，調(diào)用對應控制模塊中的處理線程，將數(shù)據(jù)提供給業(yè)務邏輯。

通過異步編程接口，傳輸框架向業(yè)務邏輯提供服務，統(tǒng)一編程接口，屏蔽數(shù)據(jù)傳輸細節(jié)。向下利用網(wǎng)絡配置文件、ActiveMQ、基于事件的異步API等，連接各個分布式系統(tǒng)，為各個系統(tǒng)提供數(shù)據(jù)通信傳輸服務，降低了各個系統(tǒng)的耦合性[5]。該框架具有較高的并發(fā)處理效率，能很好地滿足及時性處理要求。

在分布式系統(tǒng)中，接收端通過多線程、事件驅(qū)動的架構(gòu)編碼方式（Event-Driven Programming，簡稱EDP）及持久訂閱等，降低數(shù)據(jù)傳輸?shù)氖录詈闲?，降低代碼間的耦合性，保證數(shù)據(jù)及時、可靠傳輸?shù)接脩簟?/p>

2.2? 系統(tǒng)實現(xiàn)

業(yè)務層的各類程序，可通過調(diào)用框架提供的接口，將要傳輸?shù)臄?shù)據(jù)告知傳輸框架，框架將這些數(shù)據(jù)并封裝成JMS消息，發(fā)送到ActiveMQ的主題中，分布式系統(tǒng)如果已注冊監(jiān)聽該主題，則主題收到消息后將觸發(fā)接收端接收程序，數(shù)據(jù)發(fā)送到各個分布式系統(tǒng)。在此框架中，數(shù)據(jù)發(fā)送端為消息的生產(chǎn)者，接收端為消息的消費者。在系統(tǒng)中安裝了本框架，即可同時成為數(shù)據(jù)的發(fā)送者和接收者。如圖2所示。

2.2.1? 消息控制模塊設計

框架消息控制模塊負責框架初始化、消息收發(fā)、消息路由、處理優(yōu)先級控制、帶寬控制等功能。根據(jù)業(yè)務劃分不同的消息主題，代理線程根據(jù)配置文件中記錄的主題隊列、處理優(yōu)先級、帶寬等配置進行處理，代理線程運行在線程池中，將收到的消息路由到相應的業(yè)務邏輯中，業(yè)務邏輯進行接口調(diào)用后，主題線程根據(jù)配置文件，調(diào)用相應的模塊，生成消息發(fā)送到對應的隊列中。為了使配置便于擴展及解析，文件的格式選用了XML[6]。

2.2.2? 異步API

異步API實現(xiàn)了事件驅(qū)動的消息應用架構(gòu)，利用Pub/Sub（即發(fā)布/訂閱模型），使得業(yè)務邏輯與數(shù)據(jù)傳輸形成一種松耦合的關(guān)系，從而達到異步處理。且對于多個需求相同的分布式系統(tǒng)，只需調(diào)用一次API即可達到數(shù)據(jù)的多個用戶發(fā)送。

事件驅(qū)動消息應用架構(gòu)驅(qū)動著業(yè)務處理流程，框架首先進行注冊監(jiān)聽器來檢監(jiān)測消息事件。監(jiān)測到消息事件后，代理線程從消息隊列中獲取消息，將消息封裝成相應主題的主題事件，交由對應的子模塊處理，子模塊被主題事件觸發(fā)，根據(jù)主題尋找配置文件信息，與相應業(yè)務邏輯進行交互。

通過異步API屏蔽了消息等待、消息重傳、消息細節(jié)等信息，其他模塊只需針對業(yè)務邏輯進行編程即可，降低了使用難度，提高了數(shù)據(jù)處理的及時性，避免了串行處理阻塞等待的低效率。

2.2.3? 保證數(shù)據(jù)可靠傳輸

由于民航數(shù)據(jù)對于數(shù)據(jù)傳輸?shù)目煽啃砸蠛芨?，傳輸框架在發(fā)送端采用消息存儲持久化的方式，保證接收端的分布式系統(tǒng)在離線或者網(wǎng)絡不穩(wěn)定的情況下，數(shù)據(jù)也可以萬無一失地傳輸?shù)礁飨到y(tǒng)。為了降低風險，提高穩(wěn)定性，傳輸框架未選用ActiveMQ內(nèi)存對象方式保存消息，而是選用了KahaDB文件庫實現(xiàn)消息持久化。并綜合考慮服務器性能、數(shù)據(jù)量大小、頻次等因素，測試性能瓶頸，制定消息持久化的保存時間。

代碼如下：

publisher.setTimeToLive（1000*60*60*24）; // 設置有效時間

publisher.setDeliveryMode（DeliveryMode.PER SISTENT）; // 設置持久存儲消息[7]

在分布式系統(tǒng)中的消費者采用持久訂閱方式，他們分別向ActiveMQ注冊一個識別自己身份的ID（如CSAIR），當此ID離線或網(wǎng)絡不穩(wěn)定時，傳輸框架會為這個ID保存所有發(fā)送到主題的消息，當分布式系統(tǒng)再次正常連接到ActiveMQ時，該ID離線時在ActiveMQ存儲的所有消息將會被自動發(fā)送過去。傳輸框架和分布式系統(tǒng)中的消費者建立了應答機制，框架服務器在分布式系統(tǒng)中的消費者返回消息ACK標識后會更新消息的消費狀態(tài)[8]。

2.2.4? 事件驅(qū)動的消息應用架構(gòu)

在傳統(tǒng)的數(shù)據(jù)傳輸模型中，會采用定時輪詢方式來檢測數(shù)據(jù)文件是否到達，這樣的模式會導致數(shù)據(jù)處理滯后，使得框架程序需占用資源、阻塞等待數(shù)據(jù)I/O，是低效的同步處理方式。本傳輸框架利用了事件驅(qū)動的架構(gòu)編碼方式（Event-Driven Programming，簡稱EDP），在傳輸層采用注冊監(jiān)聽器的方式，當Topic中有新消息時，將馬上調(diào)用相應的業(yè)務處理接口，配合異步編程接口，減少因隨機因素造成的資源浪費，提高系統(tǒng)的性能和可伸縮性，同時保證數(shù)據(jù)到達后立即進行處理，符合民航數(shù)據(jù)時效性高的需求。

2.2.5? 服務的持續(xù)性保證探究

由于ActiveMQ支持配置集群，可以采用此方式避免服務的中斷。集群可以有效應對網(wǎng)絡中斷、服務器軟硬件故障和電力中斷等問題，與普通的備份方法相比較，集群方式使服務器切換對用戶透明，從而提供持續(xù)的服務，同時也保證了數(shù)據(jù)的可靠性和不重復性。該方式在框架中的穩(wěn)定性與可靠性正處于探究階段。

2.2.6? 框架測試

完成框架的開發(fā)和基本測試后，在民航氣象業(yè)務系統(tǒng)中使用了該框架，并聯(lián)合中國南方航空公司一起從功能、性能等方面對該框架進行了測試。測試結(jié)果表明該框架在數(shù)據(jù)可靠傳輸、網(wǎng)絡安全、縮短數(shù)據(jù)流轉(zhuǎn)時間、方便易用等方面有較好的效果。

3? 結(jié)? 論

該框架已應用于民航氣象數(shù)據(jù)傳輸，這種基于消息隊列的傳輸框架能夠在提高民航氣象數(shù)據(jù)傳輸能力的同時，保證民航氣象數(shù)據(jù)安全、可靠的傳輸，縮短了民航氣象數(shù)據(jù)與用戶的信息交換時間，提升了經(jīng)濟效益。民航各部門中存在很多可靠性和時效性要求較高的實時數(shù)據(jù)傳輸、交換業(yè)務應用場景，該框架模型在這些方面將有一定的應用前景。

在保證服務的持續(xù)性方面，可采用ActiveMQ的集群配置，進一步優(yōu)化現(xiàn)有框架，同時在高并發(fā)數(shù)據(jù)訪問時，如何通過負載均衡保證ActiveMQ服務器的服務質(zhì)量也是框架研究和改進方向之一[9]。

參考文獻：

[1] 呂常勝，張宏偉.民航氣象報文快速恢復系統(tǒng)的實現(xiàn) [J].電腦知識與技術(shù)，2019，15（2）：252-253.

[2] Snyder B，Bosanac D，Davies R. Introduction to Apache Active MQ Green Paper from Active MQ in action [M].London：Manning，2017：20-23.

[3] Snyder B，Davies R，Bosnanac D . ActiveMQ in action [J]. Manning，2011：4-5.

[4] 張煒.基于J2EE分布式架構(gòu)的高性能電商交易接入平臺研究與設計 [J].移動通信，2014，38（10）：90-96.

[5] 馮云姣，吳斌，曾輝，等.基于Active MQ的任務集成管理系統(tǒng)設計與實現(xiàn) [J].科技視界，2018（5）：39-41.

[6] 戴俊，朱曉民.基于Active MQ的異步消息總線的設計與實現(xiàn) [J].計算機系統(tǒng)應用，2010，19（8）：254-257+215.

[7] 李剛.輕量級Java EE企業(yè)應用實戰(zhàn)——Struts 2+Spring +Hibernate整合開發(fā) [M].北京：電子工業(yè)出版社，2008：185-208.

[8] 周聰.基于改進的Active MQ的通信模型的設計和實現(xiàn) [D].長春：吉林大學，2017.

[9] 龐佳麗.分布式系統(tǒng)中基于中間件的異步通信可靠性研究 [D].浙江：浙江工業(yè)大學，2017.

作者簡介：陳瑤（1987.04-），女，漢族，湖南湘潭人，工程師，碩士，研究方向：數(shù)據(jù)傳輸框架。