RabbitMQ消息队列技术在农业气象测报业务系统中的应用

成兆金，庄立伟，张媛媛，谭 奎，伍湘秦

(1.日照市气象局，山东 日照 276826； 2.国家气象中心，北京 100081； 3.华云信息技术工程有限公司，北京 100081；4.重庆市丰都县气象局，重庆 408200； 5.湖南省气象局，湖南 长沙 410000)

0 引言

中国气象局观测司2009年下发了农业气象测报业务系统(AgMODOS V1.0)，实现了观测资料的信息化，系统产生的数据文件以FTP方式传输[1]。随着自动观测技术的迅速发展，FTP文件传输方式难以满足大容量实时气象资料的高时效传输。

目前，气象通信传输多采用消息队列技术，有研究者利用Java语言，采用RabbitMQ消息队列技术，设计开发消息传输服务模块，能支持多种客户端，台站通过消息传输客户端将生产的观测数据进行消息封装后发布到消息队列(Message Queue)中，省级消息服务器从队列中获取消息后，以秒级的时间推送到国家级消息服务器，进而完成高时效的数据传输，满足农业气象观测数据不断增加和变化的需求[2-5]。

本文研究的重点是在AgMODOS的基础上，引入基于Java语言的消息服务机制，研发基层台站客户端的消息服务软件，改进新XML标准数据格式文件传输模块，进行有效的文件传输监控与管理。

1 资料和方法

消息服务客户端软件采用面向对象技术与模块化结构技术框架，利用Java语言，采用RabbitMQ消息队列等技术，运用XML可扩展标识语言开发完成[6-9]。

2 结果与分析

2.1 消息队列技术

消息(Message)是消息队列中的最小单元，实质上是一段数据，一般作为多个应用程序之间的信息传输载体，消息可非常简单，比如只包含台站要传输的XML[10]。消息队列(Message queue)，也就是消息中间件，可通过其高可靠的传递机制进行和平台无关的数据交换[11]。消息队列技术正逐渐成为各行业通信系统中广泛使用的核心技术。由于RabbitMQ采用AMQP传输协议，服务端支持消息队列的持久性，可以有效保证各个环节流转过程的观测数据完整性，同时可提供灵活的路由控制转发功能，符合气象数据传输的业务需求。

2.2 消息传输服务客户端

2.2.1 消息服务模块架构

消息传输服务模块，采用面向对象思维与模块化技术，两层架构设计，由交换微代理接口层和交换微代理服务层组成[12]。

交换微代理接口层为消息传输服务系统提供管理配置接口，接口可直接嵌入到AgMODOS中，与交换微代理服务进行信息交换。

交换微代理服务层，也就是业务逻辑层(BLL)，采用RabbitMQ消息中间件框架及common.io文件监听技术，以消息流发送的方式实现农业气象观测XML标准格式数据发送、数据接收、数据过滤/封装、数据缓存和数据补传，保证了数据内容安全完整[13]。

2.2.2 功能设计

消息收发采用消息中间件RabbitMQ，将数据按照规范的数据封装和文件名规则输出。消息服务参数采用conf.properties文件配置，完成数据接收服务器IP、发送者身份、编报中心、监视路径等消息参数配置。通过数据过滤，对数据文件的格式进行检查及重复判断，然后进行消息封装、数据发送、移盘，对发送成功的数据文件转移到Sending及sendback文件夹。发送失败，重新调用回调接口，进行数据补发，对发送成功的数据设置缓存存储，交换服务微代理功能结构如图1所示。

图1 交换服务微代理功能结构Fig.1 Exchange service micro-agent function structure

在数据接入的台站类型方面，增加区分主站、备份站的台站信息。对同一类资料，主测站和备份测站，采用不同的4级编码来区分。省级根据4级编码来区分测站类型，对主站、备份测站运行状况进行监视。

在传输服务层面，资料的格式检查增加时效性检查，对台站传输客户端的接收目录中的资料进行时效判断，对超过规定时间范围的资料数据，不再进行传输发送。

2.2.3 传输流程

台站通过消息传输客户端，对AgMODOS系统生产的XML标准格式数据Z文件进行监听、校验、封装，最后以消息的方式传给省级消息服务端，具体流程如图2所示。图中各部分运行过程：AgMODOS软件生成XML文件后写入接口目录，然后由消息传输客户端经common.io文件初始化启动监听，扫描获得文件后，进行集合唯一性重复判断，如果文件重复移盘至重复目录，通过文件名格检、fileMatch匹配，则将消息属性封装后进行消息发送至省级CTS消息传输服务器，省级不落地直接转发国家级，不通过格检的文件移盘错误目录。

(1)程序启动后监听资料入口目录，对新增实时文件或年度文件实时处理。

common.io文件监听，采用common.io.FileAlterationMonitor文件目录动态监控技术，对监控目录进行间隔1 s的实时动态监听，获取目标文件以供检查。

图2 消息传输模块流程Fig.2 Message transfer module flow chart

FileMonitor m=new FileMonitor(1 000)；//设置监控的间隔时间1 s轮询1次

//指定文件夹，添加监听

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

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

//开启监听

m.start()；

(2)通过检验规则完成对新增文件的实时检验，匹配成功的数据进行后续流程，未通过校验的文件按照相应规则进行移盘操作，并记录对应日志信息。

对于校验成功的XML标准格式数据Z文件，消息传输客户端会把客户端的配置信息和XML的属性信息(表1)封装成RabbitMQ消息属性和文件内容，形成实时消息文件或年度消息文件，封装的字节流消息体构成一个完整的消息发送给省级CTS消息服务器的RabbitMQ中，实现台站与省级、国家级CTS消息服务器的对接。

(3)对匹配成功的数据动态获取文件内容长度，并读取文件，接着将读取的XML标准格式数据内容转码为base64后存放至byte数组中封装为消息体，与消息头一起进行封装成RabbitMQ消息发送给省级CTS的RabbitMQ进行接收。

表1 农业气象观测数据消息属性Tab.1 Message properties of agricultural meteorological observation data

(4)如发送出现异常或失败时，将文件的处理信息，封装到map中，同时另开一个线程，开启定时轮询功能，以时间间隔为60 s持续从map中获取文件信息，尝试重连并发送消息，直到消息发送成功为止。文件发送成功后将文件信息从该map中移除。对发送成功的数据文件转移到Sending及sendback文件夹，如果发送失败，重新调用回调接口，进行数据补发。

2.2.4 消息服务传输操作流程

消息传输是指通过Java消息服务协议进行网络内或远程文件传输。AgMODOS系统提供简便、快捷的文件上传工具。从程序组AgMODOS中选择AgMOService(或AgMOEditor)程序项，在其主菜单上选择“XML消息传输”，调用Java程序“MsgTransferClientV1.1.jar”，启动消息服务，点击“推送”按钮，将“未传”XML文件发送至省级CTS消息服务器，立即不落地上传国家级CTS消息服务器，然后查看文件发生信息，确保消息服务正常，完成台站发报任务。如果消息服务未启动，点击“检测消息服务”按钮，检查消息服务状态，再点击“启动消息服务”按钮，启动Java消息服务，重新“推送”(图3)。

图3 消息服务传输操作流程Fig.3 Message service transfer operation flow chart

3 结论与讨论

(1)基于原农业气象测报业务系统(AgMODOS1.0)的架构技术，引入Java消息服务传输技术，弥补了FTP传输的不足。2020年12月1日的全国业务单轨运行以来，表明台站生成的XML标准文件上传及时、入库解译正确、无丢失，达到了业务运行要求，对进一步实现气象数据标准化建设具有重要意义。

(2)消息传输模块基于Java语言的jdk1.8.0版本开发，对台站计算机的操作系统要求较高，必须Win7及以上版本方可安装，对环境变量的配置必须手工操作，且Java的更新升级，必须将计算机操作系统环境变量“Java_Home”的变量值进行同步更新，更换成更新升级后的版本信息，给台站业务人员带来诸多不便，将来需进一步完善消息服务客户端软件，以适应专业版、企业版、家庭版、旗舰版等各版本的操作系统，并做到自动配置Java的系统环境变量。

(4)消息传输系统是中国气象局推行业务标准化建设新开发的业务系统，是一个新概念，要求台站业务人员认知消息服务的概念和有关知识。业务准入运行以来：系统运行稳定，Z文件数据99%在1 s内完成从台站到国家局的传输，达到了业务运行要求。