智能网格气象产品共享管理系统的设计与实现

周雪莹，胡丽丽，魏敏丹

（江西省气象信息中心，江西 南昌 330096）

智能网格气象产品定量化和精细化是气象业务发展的必然趋势，对公众和各行业领域的精细化天气预报服务具有重要的作用[1-2]。随着智能网格气象业务不断发展，省级部门接收、制作和发布的网格产品数据量日益增加，频繁操作产品文件会影响服务器的整体性能，导致文件不能及时下载使用等问题，为了保障智能网格气象产品高效共享应用，针对目前的文件共享方式的缺点，该文设计开发了一套智能网格气象产品共享应用系统，实现消息生产和消费、异步传送以及实时监听等功能。

1 消息传输流程设计

采用开源消息代理软件，通过服务端和客户端的方式，实现客户端实时获取服务端授权订阅的产品文件状态，通过有效的产品文件下载服务实现产品文件实时同步共享。当有新的产品文件到达省级数据环境或者被监听的文件目录有文件变动时，触发监视模块的实时监视代理，生成标准格式的产品文件订阅消息，通过处理线程池发送到开源消息代理软件[3-5]，客户端收到订阅的消息后，启动文件同步模块实现远程文件下载，下载结束后客户端把下载情况生成状态消息发送到开源消息代理软件，如果下载过程中有异常情况，如产品文件下载不完整、服务器状态不正常等，通过异常处理模块发送到监控系统。消息传输流程设计图如图1 所示。

图1 消息传输流程设计图

2 总体设计

智能网格气象产品共享应用系统总体设计包括数据源区、功能模块区和告警区。系统总体架构图如图2 所示。

图2 系统总体架构图

数据源区主要是通过监视模块实时监视从各种渠道收集文件的服务器群，如果有新增、修改、移动等文件操作时，生成数据消息并写入日志文件，同时把产品文件的基本信息保存到业务数据库。功能模块区处理接收到的消息，开源消息代理软件会把有订阅需求的消息发送到消息队列中[6-9]，客户端根据接收的数据消息下载数据文件，同时把消息内容和出错记录写入日志文件，并且根据业务或科研需要对下载的产品文件进行保存，为了防止出现僵尸进程或者进程异常退出，导致整个系统运行不稳定，建立了一套简单的机制以保证进程一直处于活跃状态[10]。告警区主要通过异常处理模块把进程运行、服务器性能或同步情况等异常信息，通过短信、微信或页面的形式发送给值班业务人员或管理人员。

3 系统设计与实现

在国内气象通信系统（CTS2.0）和虚拟资源池的基础上，以收集到的各类气象网格产品文件为数据源，该系统采用开源消息代理软件RabbitMQ、MySQL数据库、C#和Python 等技术，该系统实现消息生产和消费、异步传送和实时监听等功能。

3.1 监视代理

监视代理包括实时监视代理和状态监视代理两部分，分布式部署在服务器上，收集文件路径、大小和操作等基本信息，以及进程状态等可用性数据，并且通过守护进程保证进程一直处于运行状态。

实时监视代理是基于Python 的实时监控第三方库watchdog 进行开发的，实时监听代理的工作原理如图3 所示。

图3 实时监听代理工作原理

通过参数设置控制监视策略，如指定监视的文件类型、是否区分文件名大小写、是否监视子文件夹，并且采取多次调用schedule 方法的方式，同时监视多个文件夹，通过操作系统的时间触发，不需要循环和等待，当共享清单下的特定文件被创建、删除、修改、移动时，生成数据消息并与订阅清单进行对比，如果消息的内容符合订阅清单的要求，则把消息发送到开源消息代理软件中对应的消息队列。

3.2 消息处理

消息处理软件包括消息封装模块和消息处理模块，分别部署在服务端和客户端。消息封装模块是把接收到的产品文件信息根据配置文件进行标准封装，使用开源消息代理软件RabbitMQ 的异步消息队列（Work Queue）方式投递消息，不需等待客户端响应可直接将消息来源、发送时间和发送目的地等信息写入日志文件。消息的传输格式是以键值对的形式存放消息属性，包含消息发布者信息、消息接收者信息、消息发送时间、文件大小和文件绝对路径等。

消息处理模块部署在客户端，客户端从消息队列中读取订阅的消息，根据获得的消息属性调用文件同步中的文件下载模块，并把下载文件的基本信息写入日志文件，如文件名、文件大小、保存路径和来源等，同时存入关系数据库MySQL 中进行持久化存储。

3.3 存储管理

该系统采用关系型数据库管理系统MySQL 来存储信息，为了提高数据读写性能、数据库访问速度和数据查询速度，保证数据入库的及时性，采用视图、索引、主键等对数据库进行优化设计[11-13]。根据业务需求，系统设计了服务器状态表、文件详细信息表、文件总量数据表、异常情况数据表和下载情况数据表等。服务器状态表用于存储监控区域服务器和客户端的系统进程、IO、系统时间和磁盘等出现异常的信息；文件详细信息表用于存储监控区域共享文件的详细信息；文件总量数据表用于存储监控区域共享文件每日下载的数据总量；异常情况数据表用于存储读取消息过程或下载过程出现的异常情况；下载情况数据表用于存储客户端下载数据的文件名、数据量、时间和目录等信息。

3.4 异常处理

异常情况包括文件同步失败、服务器状态异常和进程异常等，异常处理模块采用Web 页面的方式，提供阈值设置和管理、告警方式的选择和异常记录历史查询等功能。告警阈值设置有3 个级别，根据不同的阈值触发不同级别的告警[14-16]；告警方式有微信、短信和Web 页面3 种方式，根据值班业务人员或管理人员的需求选择一种或多种方式进行告警；异常记录历史查询可以查询所有的异常记录以及处理结果。

4 关键技术

4.1 基于RabbitMQ的异步消息传输

RabbitMQ 是基于高级消息队列协议(Advanced Message Queuing Protocol,AMQP)的开源消息代理软件，设置队列中的消息持久化，用来处理实时且要求高可靠性的消息。该文采用RabbitMQ 的主题交换器（Topic Exchange），使用特殊含义的符号为路由器关键字（Routing Key）设置特定规则，根据路由器关键字发送消息到所有使用匹配绑定键绑定的消息队列（Queue），如果不存在路由器关键字中指定的消息队列名，则指定到这个消息队列的消息会被丢弃。客户端使用basic.consume 命令，订阅消息队列中的消息,实现信道（channel）自动在处理完上一条消息之后，接收下一条消息[9-10]。

消息按照主题方式发送，消息队列名称以“Q.”开头，包含资料来源和资料类型，主题交换器的路由器关键字是由点分隔的一组单词组成，“#”表示0 个或若干个关键字，“*”表示一个关键字，将消息发送到一个或者多个消息队列中，实现pub/sub 模式，即发布订阅方式。主题方式消息传输图如图4 所示。

图4 主题方式消息传输图

4.2 文件系统实时监听

基于Python 的实时监控第三方库watchdog 优先使用底层原生API，其次再通过轮询磁盘实现监控。通过操作系统的时间触发，不需要循环和等待，对不同平台的事件都进行了封装，通过observer 模块不断检测调用平台的依赖代码，监控指定目录/文件，当有创建、删除、修改、移动等变化时，产生事件并且对应特定的事件类，再通过事件处理类（event_handler）来处理对应的事件，observer 模块、事件处理类和被监控的文件夹通过observer.schedule函数串联起来。

文件系统实时监听调用文件系统事件处理类，通过定义处理事件，继承FileMovedEvent、FileCreated Event 和FileModifiedEvent，并重写对应实例方法，来实现文件实时监视。文件写入目录的过程中会多次触发类实例self.on_moved(event)，为了避免文件重复处理产生重复记录，导致大量相同的消息发送到消息代理软件，把文件名赋值给一个全局变量，如果上次传入的文件名和这次赋值的文件名是一致的，则不做任何操作，从而避免多次触发类实例。为了只对格点资料进行监视控制，使用PatternMatching EventHandler 事件处理类，指定模式为GRB、GRB2、MIC 和BIN，如果监视目录下的文件名匹配指定的文件类型则调用实例方法。

4.3 客户端状态信息定时采集

采用基于Quartz的Python定时任务框架apscheduler和系统工具模块psutil 实现定时采集客户端进程运行状态信息的任务。psutil 是一个开源且跨平台的库，提供了便利的函数，用于获取系统运行的进程和系统利用率等信息。操作系统基本都会提供定时任务的实现，例如Linux 自带的crontab、Windows 自带的任务计划，但是apscheduler 提供了构建专用调度器或调度服务的基础模块，以及非常丰富而且方便易用的定时任务接口，提供了基于日期、固定时间间隔以及crontab 类型的任务，能实现更好、更便捷的跨平台定时任务。客户端状态信息定时采集时，采用后台运行（BackgroundScheduler）的方式，每隔5 min采集一次进程信息并写入到业务数据库中。

5 结束语

该文在虚拟资源池的环境下，设计了具有消息生产和消费、异步传送、实时监听和文件远程备份等功能的智能网格气象产品共享应用系统，并且对消息生成、消息处理、实时监视和文件同步进行了详细的阐述。采用RabbitMQ开源消息代理软件，实现消息异步处理，实时生成产品文件的数据消息，并且发送到消息队列，客户端根据接收的数据消息下载对应的产品文件，减少了服务器内存的使用量，提高了数据共享时效；利用差异备份的方式备份变动的数据，减少宽带资源的大量耗费，提高数据传输效率；定时采集客户端状态信息获取系统运行的进程和系统利用率，便于监控系统和管理进程，保证系统安全可靠运行。