基于数据流的无人值守泵站群物联网平台设计

岳敏

（重庆远通电子技术开发有限公司，重庆 400000）

1 背景

在供水与排水网中，有大量的无人值守泵站，7×24小时工作。为了保证这些泵站的正常工作，建立了联网监控系统。这些系统主要采用比较传统的自动化技术构建，包括使用组态软件作为现场数据采集的核心软件、使用Modbus 作为现场数据采集及向中心站的数据传输协议、使用关系型数据库存储工况数据，如图1 所示。

图1 以组态软件为核心的监控平台架构

在应用中，该方案虽然解决了无人泵站的远方监控问题，但也表现出较多的问题，主要表现在以下方面。

（1）软件架构方面。组态软件因架构限制，阻碍了监控系统的智能化拓展。因应用需求，需在基础监控之上进一步开发泵站能耗分析、故障诊断等智能应用。但组态软件架构比较封闭，提供的开放集成接口较少，既不能有效提供数据、分析技术等方面的基础支持，也不能方便地将智能应用集成进组态应用中，已不能满足智能应用开发需求，从整体软件架构上，阻碍了整个系统的智能化拓展。

（2）网络传输协议方面。Modbus 协议在安全、功能、效率等多方面已不能满足需求。Modbus 作为应用最广泛的工控网络协议，在现场设备数据采集方面具有最好的兼容性，具有不可替代的价值。但其协议几乎不具备安全机制，因此，在监控站向中心站的广域数据传输方面有严重的安全隐患。同时在新开发计划中，拟进一步建立分布式的设备群事件机制，开发泵站的全生命周期管理应用，Modbus 的轮询模式从功能性上来说无法支持这些新功能的开发，并在监控站数量不断增加的情况下，出现了明显的效率瓶颈。

为解决以上问题，以面向大量无人值守泵站群为目标，采用新的物联网技术及IT 技术，设计实现了新的无人值守泵站群物联网平台。以下将从平台架构、协议设计、应用接口等方面对新的设计进行详述。

2 架构设计

以数据流处理为基本设计思想，采用MQTT 为通信协议，各单元间以标准的集成方式进行集成的新架构，如图2 所示。

图2 以数据流处理为核心的系统架构

2.1 基于MQTT 的通信架构

为解决Modbus 协议带来的诸多架构局限，采用了以MQTT 为通信协议的通信架构。MQTT 是目前物联网系统中应用最广的协议，订阅/发布是MQTT 的基本通讯模式。在该模式下，云中心通过订阅设备端的相关消息，实现设备端向云中心的数据发送；同时，设备端通过订阅云中心的相关消息，实现接收云中心的控制命令。订阅/发布模式在大量远方设备情况下，相对轮询模式具有明显的性能优势，节约大量的网络资源。

2.2 基于数据流的设计思想

平台架构以处理数据流作为基本思想。数据流的主要特征是数据处理单元将数据看作无边界的数据集合，在数据传输的过程中即完成数据转发、过滤、分析、转存等操作，相对于将数据按时间间隔存储后再进行处理的批处理系统，数据流系统具有处理延迟小、计算负荷分配均匀（从而更适合规模扩展）等特点。Apache Storm、Apache Flink 等流处理系统的广泛应用，已证明了数据流处理系统在数据密集应用中的优越性和实用性。

在泵站监控中，主要包括三种数据流：（1）设备数据流：即设备的工况数据，是典型时间序列数据；（2）事件数据流：即监控系统中常见的报警事件、越限事件等事件数据。有效设计、处理这类数据，形成事件驱动体系，是提升系统智能性的关键之一；（3）控制数据流：主要为云中心对监控站的控制信号，也包括中心对现场系统的运维信号，例如，设备升级数据包、远方提取日志命令等。

针对不同数据流的结构特征差异，平台使用了混合的数据存储方案：对于工况数据这类时间序列数据，使用时间序列数据库存储，例如，实际采用了TDEngine；事件数据因具有比较明确的结构特征，仍采用关系型数据库进行存储，例如，采用了PostgreSQL。

在监控站端使用LF EDGE ekuiper 作为流处理核心。LF EDGE 是Linux 基金会发起的开源组织，旨在建立独立于硬件、芯片、云或操作系统的一个开放的、可互操作的边缘计算框架。作为LF EDGE 下孵化的项目，ekuiper 是一款轻量级物联网边缘分析、流式处理开源软件，可以运行在各类资源受限的边缘设备上。eKuiper 参考了Apache Spark、Apahce Storm 等云端流式处理项目的架构与实现，结合边缘流式数据处理的特点，采用了编写基于源（Source），SQL(业务逻辑处理)，目标（Sink）的规则引擎来实现边缘端的流式数据处理。ekuiper 的基本架构如图3 所示。

图3 ekuiper 的基本架构

监控站的数采单元通过modbus 协议采集二次供水泵站PLC 和其他传感器的数据，并将数据转换为MQTT 数据包，发布到监控站的MQTT broker（使用Mosquitto）。ekuiper 通过向Mosquitto 订阅这些数据，获得Mosquitto 转发的设备数据流。根据规则设定，ekuiper 将一部分需要在云中心即时显示的数据，直接通过MQTT 转发布到云中心的MQTT Broker，例如，泵站实时出水压力、实时功率等；将不需要在云中心实时展示的数据，例如，变频器频率、电机三相电流等，存储在本地，供本地智能应用分析使用。

监控站各单元产生的事件，也通过ekuiper 汇总后发送给云中心的MQTT Broker，构成事件数据流。ekuiper 通过MQTT 接收云中心发送来的控制数据流，并根据规则转发给特定单元，执行相应控制命令。

基于MQTT 的消息命名机制，实现了对不同数据流的分类标识，具体设计将在第3 节中详述。云中心接收到监控端的数据流后，流处理引擎根据分流标识对数据进行分流：工况数据流将存入时间序列数据库；事件数据流将存入关系数据库。

2.3 基于标准接口的系统集成

一方面，监控应用需求不断扩展，为因应需求变化，新增或更新相应智能应用，应是平台在生命周期内的常态化要求。另一方面，IT 技术发展迅速，如何将最适合的IT 技术不断引入平台内赋能，是不断拓展平台功能，提升平台性能的重要保证。因此，保证单元接口的标准化是平台设计的重要原则。

通过对技术体系现状的分析，确定了两个层面的集成接口机制：

（1） 基 于Restful API 的 服 务 集 成。Restful API 是目前使用最为广泛的服务集成方式。通过使用Restful API 能够实现微服务架构，解耦服务间的紧耦合，提升平台的可扩展性。平台集成的几个核心模块，例如，ekuiper、TDengine 均提供了Restful API，以实现对其功能的调用。在本平台中，目前主要在两种场景中使用Restful API 集成。第一种是在集成调用以上核心模块的功能时。第二种是在作为控制单元的控制台（包括监控站与云中心的控制台）对外提供对系统的操作、配置、展示等功能时，均以Restful API 方式提供。通过该方式可实现控制台核心控制逻辑与使用方式的解耦，例如，当实现控制台界面时，可以灵活选择各种不同的前端技术（例如基于浏览器的Web UI，或者基于桌面的UI），还可以通过脚本实现对系统的自动化运维。

（2）基于ZeroMQ 的操作系统内进程间通信。当实现同一操作系统内不同智能应用间的交互时，例如，故障诊断应用调用能耗分析应用的计算结果作为其故障诊断的依据时，采用基于ZeroMQ 的操作系统内进程间通信方式可获得更高的性能。ZeroMQ 是一种基于消息队列的多线程底层网络库，支持线程间，进程间、机器间的消息通讯，提供线程安全的同步接口。

3 协议设计

在2.2 节中已提到将数据流根据其性质进行分流是平台的核心思想。在具体实现上，主要依赖对MQTT 数据包的主题（Topic）设计，以及流处理引擎对主题的解析进而进行对不同数据流的分流。每个MQTT 数据包包括主题与数据体(Payload)两部分。MQTT 规范除了规定#与*两个通配符外，并未对主题的设计做出其他规定。因此，利用这种灵活性，设计了如下的主题结构：

应用名称/设备名称/数据流类型/…

应用名称是指整个平台的名称，例如，对于二次供水监控平台，采用拼音首字母命名模式，命名为“eg”。设备名称指具体一台二次供水设备的编号，例如，可命名为“d101”，其中101 是设备编号，在前面加入字母是为未来在该应用平台中加入其他类型设备预留空间。数据流类型即为以上所述的数据流类型，包括设备数据流（d）、事件数据流（e）、控制数据流（c）。

基于以上规则，对于编号为101 的二次供水泵站的MQTT 数据包主题可命名如下：

（1）工况数据，“eg/d101/d”，具体的数据格式在数据体中采用JSON 格式表达。

（2）事件数据，“eg/d101/e/event_name”，其中“event_name”为具体的事件名称，例如，“online”表示设备上线事件；“offline”表示设备下线事件，跟事件相关的数据包含在数据体中，例如，“eg/d101/e/online”事件的数据体中可包括系统的开机自检报告。

（3）控制数据，“eg/d101/c/command_name”，其中“command_name”为具体的控制命令，例如，“reboot”表示设备重启命令，收到“eg/d101/c/reboot”命令后，101 号设备应自启。

从以上可见，该命名方式具有极大的可扩展空间，能够胜任平台未来不断扩展的需要。流处理引擎在接收到MQTT 消息后，将首先对主题按照命名规则进行分析，分析出数据包的来源设备，数据流的类型，并根据不同类型数据流的特定处理方式对数据进行处理，实现数据流的分流。

4 应用与展望

目前，该平台已接入超过100 台二次供水设备。在实现基础的远程监控功能基础上，进一步支撑了智能应用的快速开发和可靠运行，包括实现了发明专利（CN202010099925.1）《一种边云协同的供水设备能耗监测与能效评价系统和方法》中的二次供水设备能效评价算法，为有效降低二次供水设备能耗奠定了数据基础。未来，平台将进一步加大二次供水设备接入量，并通过核心单元集群配置、云原生部署等方式提升平台的可靠性和管理效率。