消息队列技术在气象软件系统中的应用

■ 王恩文

国内气象通信软件系统第二版（CTS2）中的消息传输模块已经在全国2000多个气象站进行业务运行。目前全国所有的国家级地面自动站BUFR数据均通过该系统进入国家级、省级的气象数据环境，并且为省级的MDOS系统提供了数据服务支撑。从试点运行到现在的业务运行，消息传输运行稳健，90%的数据从台站到国家的传输时间在2 s以内，99.99%的数据在10 s以内完成传输。

国内气象通信系统是承担国内气象资料和产品收集以及国内外气象资料和产品的国内分发的业务系统。随着现代气象业务快速发展，气象数据的种类、数量日益增加，关键业务资料观测、更新频次达到分钟级，传输时效要求达到秒级。现有国内气象通信系统，包括目前业务运行的新一代国内通信系统、国内气象通信软件系统第一版（CMA domestic telecommunication system（version 1），CTS1），仅支持国家级和省级间基于文件的数据传输，难以满足未来的统一省以下通信传输技术体制，难以支持多种类、高频次、大容量实时气象资料的高时效传输。

国内气象通信软件系统第二版（CMA domestic telecommunication system（version 2），CTS2），在现有系统基础上新增消息传输、流传输模式，并以Redis消息队列（Redis message queue，RMQ）为核心的交换控制系统替换现有的调度系统，大幅度提升了气象数据实时传输效率。

1 消息队列技术

1.1 消息队列基本概念

消息（Message）是消息队列中最小的概念，本质上是一段数据，它能被一个或者多个应用程序所理解，是应用程序之间传递的信息载体，消息可以非常简单，比如只包含文本字符串、XML、JSON等，也可以非常复杂，比如图片、BUFR（Binary universal form for representation of meteorological data）编码的二进制文件。

消息队列（Message queue，MQ），也可以称为消息队列中间件或消息中间件，是指利用高效可靠的消息传输机制进行与平台无关的数据交流，并给予数据通信进行分部署系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间通信。

消息队列一般有两种传递模式：点对点（P2P，Point-to-Point）模式和发布/订阅（Pub/Sub）模式。点对点模式是基于队列的，消息生产者发布消息到对垒，消息消费者从队列中接收消息，队列的存在使得消息的异步传输成为可能。发布/订阅模式定义了如何向一个内容节点发布和订阅消息，这个内容节点称为主题（Topic），主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题， 而消息订阅者则从主题中订阅消息。主题使得消息的订阅者与消息的发布者保持独立，不需要进行接触即可保证消息的传递，发布/订阅模式在消息的一对多广播时采用。

1.2 采用RabbitMQ的原因

目前市场上的消息中间件有很多，比较主流的消息队列有RabbitMQ、ActiveMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ。

在系统设计初期进行了大量的业务环节消息中间件选型测试，从以下方面进行选型和对比：

1)传输效率：在所选的6类消息MQ中传输效率ZeroMQ最高，RabbitMQ次之。

2）持久化消息：ZeroMQ不支持，ActiveMQ和RabbitMQ都支持。持久化消息主要是指机器在不可抗力因素等情况下挂掉后，消息不会丢失的机制。

3）综合技术实现：可靠性、灵活的路由、集群、事务、高可用的队列、消息排序、问题追踪、可视化管理工具、插件系统等。RabbitMQ/ Kafka 最好，ActiveMQ次之，ZeroMQ最差。当然ZeroMQ也可以做到，不过必须手动写代码实现，代码量不小。

4）高并发：RabbitMQ最高，因为它的实现语言是天生具备高并发、高可用的erlang语言。

经过综合考虑，RabbitMQ最符合业务需求。

1.3 消息队列的作用

消息队列凭借其独到的特性，在不同的应用场景下可以展现不同的作用：

解耦：消息队列在处理过程中间插入了一个隐含的基于数据的接口层，这允许程序独立地扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束即可。CTS2中消息通信解耦了数据传输线路中各环节的生成、入库等应用的强关联，各应用只须做好自己的工作，无须担心向下游的程序是否正常，简化了系统架构。

冗余（持久化）：处理数据有时进程失败。除非数据持续存在，否则它将永远丢失。消息队列通过持久化数据，直到完全处理来缓解这一点。许多消息队列使用的put-get-delete范例需要一个进程来明确指出在邮件从队列中删除之前已经完成了处理消息，确保数据保持安全，直到完成。应用场景：CTS2中正是使用这一特性来保证数据的完整性。

可扩展性：消息队列解耦了应用的处理过程，所以可以轻松地扩展。将消息添加到队列，只须添加另一个进程。不需要更改代码，不需要调整任何配置。应用场景：CTS2中利用这一特性，可使国家级、省级各个平台根据直接的数据规模配置适量的消息队列服务器（Broker）即可。

弹性（流控）：消息队列能够使关键主键支撑突发访问压力，不会因为突发的超负荷请求而完全崩溃。如果正在处理来自队列的消息的进程失败，那么仍然可以将消息添加到队列中以在系统恢复时进行处理。应用场景：在CTS2中这一特性可以使当数据源端异常时（比如传输客户端因配置错误突然导入多年历史数据），整个传输系统可以保证稳定运行。

缓冲：在任何重要的系统中，都需要不同处理时间的组件。消息队列通过提供缓冲层来帮助这些任务以最高效率运行，该缓冲区有助于控制和优化数据流经系统的速度。应用场景：CTS2系统中数据源程序、数据处理程序、数据入库程序无须保证同时启动，任何环节的应用再启动后只要有任务或数据即可进入工作状态。

异步通信：消息队列启用异步处理，允许将消息放在队列上，而不是立即处理。可以随意添加多个消息，然后随意处理。这样可使CTS2在下游系统未启动的情况下，仍然正常收发。

2 RMQ在CTS2中的应用

在CTS2中，RMQ消息传输承担着全局任务管理、数据收集任务管理、数据分发任务管理、消息数据传输的总要任务（图1），是国家级地面自动站BUFR格式消息数据的唯一传输途径，是全局任务管理的核心。

图1 消息队列在CTS2系统架构中的应用

2.1 消息传输架构设计

国内气象通信系统基于RabbitMQ开发传输客户端和消息服务端，其中传输客户端部署到台站，消息服务端部署到省级和国家级，传输流程如图2所示（以地面自动站BUFR格式数据为例）。

2.1.1 消息数据传输基本架构

国家级地面自动站BUFR格式消息数据传输流程：

1）传输客户端与ISOS实现消息API接口和文件目录接口，将获取的消息数据上传至省级消息服务端。

2）省级MDOS（Meterological Data Operational System）调用消息接口从消息服务端获取消息数据，并将质控后的消息数据送至消息服务端，消息数据入省级数据环境，消息收发处理进程从省级数据缓存获取消息数据，导出成文件，同时在省级通过配置shovel，将消息数据shovel到国家级RabbitMQ队列。

3）消息数据入国家级数据环境。

消息数据传输主要包括台站数据上行和国家级控制指令下行。

全国基础气象台站的各种观测资料通过传输客户端将数据传输至本省省级通信系统，省级中心将省（区、市）的全部观测资料、加工产品以及其他有关信息上传至国家级通信系统。

图2 数据传输流程图

国家级业务系统由国家级传输客户端发送下行指令，通过国家级、省级通信系统，由台站级传输客户端提供给台站综合观测业务系统。台站业务系统响应指令后，通过台站传输客户端发送上行响应，国家级通信系统通过国家级传输客户端向该国家级业务系统进行反馈，同时抄送国家级和省级监视系统（客户端）。

图3 物理部署图

图4 数据传输时效分析图

省级业务系统由省级传输客户端发送下行指令，通过省级通信系统，由台站级传输客户端提供给台站综合观测业务系统。台站业务系统响应指令后，通过台站传输客户端发送上行响应，省级通信系统通过省级传输客户端向该省级业务系统进行反馈，同时抄送国家级和省级监视系统（客户端）。

2.1.2 高时效设计

为了满足国家级地面自动站BUFR数据的秒级传输及以后系统消息传输数据量可扩展的需求，设计了RMQ多机多节点的集群模式（图3）。

图4为双机多节点的数据传输时效性测试图。通过大量的模拟和业务测试绘制出时效性曲线。可以看出，在所实施的测试条件下，无论每台机器部署几个节点，都能很好地满足传输时效性小于2 s的要求。若数据的传输时效性要求小于1 s，从测试曲线看出，每台机器部署8个节点，同时部署2台服务器即可满足地面自动站BUFR数据的秒级传输要求，如需要对业务进行扩展，只需成对增加服务即可。

2.1.3 安全可靠性设计

RMQ多节点多集群模式除了可以大幅度提升传输效率，还可以允许发布者与生产者在单节点崩溃的情况下继续运行，当集群内失去一个RMQ节点时，客户端能重新连接到集群的任何其他节点并继续生产或消费。

另外，在传输客户端进行生产（发送数据）或消费（接收数据）时，分别采用了发布确认模式和接收确认模式。发布确认模式可以保障消息到达消息服务器，当指定时间内无法到达消息服务器会返回发送超时代码，客户端可以尝试进行重新发送。接收确认模式在消费者接收消息后消息体仍然存在于消息队列中并且被标识为“Unacked”，直到消费者处理完消息后发送“ack”指令，消息体才会从队列中删除，如果消费者异常崩溃消息会重新恢复正常状态供其他消费者消费，从而保证消息不丢失。

图5 交换控制流程图

2.2 交换控制

交换控制配置项包括收集任务管理、分发任务管理、编辑任务管理和元数据同步任务管理四个部件（图5）。各任务管理部件实时读取内存中的策略配置，解析其中的时间策略和操作环节，根据时间策略触发生成作业任务，并按照任务优先级将作业送入到相应消息队列中。相关配置项从对应消息队列中获取作业任务，解析任务策略并调度任务执行，当前环节操作完成后，将按照后续流程将数据对象向下游传递。

3 结语

CTS2中的消息传输模块已经在全国31个省（区、市）的2000多个国家站进行业务运行，目前全国所有的国家级地面自动站BUFR数据均通过该系统进入国省的气象数据环境，并且为省级的MDOS系统提供了数据服务支撑。从试点运行到现在的业务运行，消息传输运行稳健，90%的数据从台站到国家的传输时间在2 s以内，99.99%的数据在10 s以内完成传输。

CTS2中的交换控制模块也在国家级、贵州省、湖北省开展了试点运行，承担着3个试验点的文件传输、消息传输、流传输的调度工作，到目前为止运行稳定，无重大故障发生。

深入阅读

陈玉华, 翟颖佳, 周红, 等, 2013. 基于气象通信系统切换信息传输流程梳理与优化. 农业网络信息, (7): 22-24.

林润生, 孙周军, 谭小华, 等, 2011. 新一代国内气象通信系统设计与实现. 气象, 37(3): 356-362.

刘金霞, 王慧瑜, 赵威, 等, 2012. 省级新一代气象通信系统传输流程的设计与实现. 气象与环境学报, 28(6): 76-80.

向筱铭, 徐晓莉, 宋智, 等, 2017. 基于CTS的台站上行气象数据传输监控平台的设计与实现. 气象科技, 45(4): 647-652.

张来恩, 王鹏, 韩鑫强, 2018. CTS2.0消息封装及交换控制策略设计及实践. 气象科技进展, 8(1): 271-273.