《2018年中国水资源公报》显示，为实现水资源质量评估，全国需要监控124个湖泊、1 129个水库、6 779个水功能区、2 833个地下水监测井、1 045个集中式饮用水水源地进行采样调查。以北京市为例，2018年全市地表水资源质量监测站点共221个，监测河段104个，湖泊22个，大中型水库18座。监测项目共44项，年监测频率12次。

各省市具体会针对水资源进行集中评估，但并非实时记录评估。以目前国内水资源信息公开的情况来看，数据集中以月度、季度等方式定期记录与发布，地表水水质实时监测数据仅有中国环境监测总站发布的“国家地表水水质自动监测实时数据发布系统”可供查询，系统中的数据也并非完全实时，而是每隔数小时记录监测点在整点监测的数据。

因此，只有实时监测，才能科学、准确地进行资源配置及调度、对环境质量进行动态评价和有效监督、应对可能发生的水污染突发事件，保证供水安全。因此在建立供水监管平台的过程中，实现水质数据的实时监测是必不可少的环节。

在供水全过程监管平台中，实时监控数据可应用在多个场景中。

(1）水质告警。水质告警是水质监控过程中的重要一环，使得应对机制和回溯机制得以发挥作用，一方面实现对污染地区的及时处理、对具有污染风险地区的预告和预案准备；另一方面作为排查污染源和污染物的关键记录。

(2）水质分析。分析手段包括历史水质数据分析和与水质督查、上报数据的交叉对比分析，基于数学模型，可以得出水质变化情况和数据的准确程度，结合其他社科数据，得出与气候、季节等自然因素及人口、经济等社会因素相关的结论，作为水文研究的重要依据。

(3）水质决策。实时数据可以为政府部门制定水资源保护法规、标准和规划，开展水环境保护管理工作以及科学研究提供数据，为污染源管理和排水的奖惩机制提供依据。目前，各地对水质均有采样分析过程，并积累了大量数据，但缺乏统一的分析管理机制和平台。面对数量庞大、来源众多的水质数据，平台希望实现对数据的统一接入、分析、管理功能。本文设计的水质实时监测系统实现的目标包括：(1)支持多监测设备或监测平台接入；(2)支持海量水质数据接入，保证实时准确；(3)实现水质数据的实时展示；(4)统一水质标准，实现水质超标告警；(5)系统实现目标需要满足的特性包括高并发、高实时性、高性能。在本文设计系统中，引入了几种技术来实现以上功能。

Netty是一个高性能、异步事件驱动的NIO网络通信框架，支持多种编解码功能，支持海量并发，具有高扩展性和高可用性的特性，已经在各大企业中得到应用。水质实时监控系统选用Netty作为通信框架，选择主从Reactor多线程模型，可以同时支持成千上万个客户端的连接，满足高并发、高负载、传输快的性能要求。Netty可以实现多连接高并发，在实际应用中，能够保证各地市的多个采样设备或平台同时接入、同时完成数据传输，避免因采样设备过多产生的通信问题。

水质实时监测连接的客户端多，产生的实时数据量大，峰值处理能力存在一定的问题，利用Kafka高吞吐量、低延迟的特性，可以很好地解决通信链路和实时数据推送问题，有助于控制和优化数据流经过系统的速度。消息队列令数据处理保持稳定的高性能，能够保证采集数据上报过程不会因系统性能原因产生数据丢失等意外，同时避免了因各个采样设备或平台上报频率不同，在数据上报高峰期及上报低谷期系统资源不平衡，保证在不同频率场景中，用户都能够及时准确地接收水质数据。

为了让用户能够查看最新实时监测数据和水质的变化情况，服务器端利用Websocket通信技术主动将接收到的监测数据推送给用户，使看到的数据动态自动刷新。实时推送保证用户在第一时间获取水质信息及报警情况，从而及时发现污染等异常情况并制定应对政策，将水质异常状况造成的环境问题和经济损失降到最低。

本系统采用TCP/IP协议来完成终端设备与服务器端的连接，TCP/IP协议是可靠的字节流协议，每帧数据量大小不是固定的，会存在半包、粘包问题，导致服务器端接收到的数据不完整，因此需要用自定义的通信协议来发送和接收数据。本系统中使用《污染物在线监控（监测）系统数据传输标准》（HJ212-2017）的数据传输协议，支持环保212数据传输标准，能够完成与环保局以及各类终端系统平台的实时水质在线监测数据对接。

数据传输协议统一了数据格式，解决了数据难以汇集的问题。由于采集设备或平台由各地自行设计实现，往往出现各自为战、数据不统一的情况。国家统一供水监管平台能够通过数据传输协议规定数据内容（包含统一的水质指标检测要求）及数据大小，保证监管数据完整性的同时，平衡各地数据上报的网络开销及资源消耗。

本系统使用到的开源技术主要有Netty通信框架、Java语言、Oracle数据库、Websocket通信协议、Redis数据库和Kafka消息队列。本系统主要由Netty服务器端和web服务器端2个部分组成。设计的水质实时监控系统的整体架构如图1所示。

系统对实时水质监控数据的处理包括4个步骤。

(1）采集设备接入。采集设备以终端传感器、上位机、数据汇总平台等不同形式，将水质数据以采集状态或汇集封装后的状态上传至系统，上传的前提是采集设备与系统建立安全、可靠的网络连接。由图1的系统架构情况可知，采集设备将与系统中的Netty服务器端建立TCP长连接。TCP长连接的特点在于连接双方会主动采用心跳机制维持双方连接，考虑到水质实时监控的应用场景，数据传输具有持续性，长连接可以避免重新建立连接过程中的网络开销和本地资源浪费。目前单台Netty服务器可以支持上万的长连接，通过分布式部署等方式，系统支持的接入设备数量可以不断扩展。

(2）实时数据传输及解析。采集设备接入后，可以通过Netty实时传输数据。为了便于数据的拆分和解析，系统采用了HJ212通信协议，要求以规定的数据格式传输数据。HJ212协议本身具有一定的通用性和标准性，Netty服务器接收数据后会转发给自定义的解码器，解码器将数据解析成为JSON格式。HJ212协议能够在保证数据的准确性的同时，维持了数据内容的轻量级，提升处理性能和系统吞吐量。

(3）实时数据推送。解析完成的数据需要在供水监管平台页面中实时动态展示，同时也可以实现与大屏等其他可视化平台的对接。系统将通过Websocket通信协议，将实时数据广播，广播范围由实时数据来源的采样点（采集设备）及其所在行政区决定，不同用户可以接收不同的实时数据；实时页面通过订阅广播、实时监听的方式实现数据的即时更新。

(4）实时数据业务处理。这里的业务处理主要指水质指标的超标判断及告警发布。目前系统主要依据监管平台指定的水质标准进行超标判断，过程包括水质指标提取、规则解析、监测值判定等，业务逻辑较为复杂，耗时较长。因此系统引入了消息队列和规则引擎。消息队列将数据集中后依序处理，一方面防止数据高峰期的数据堵塞、减少系统延时；另一方面充分利用数据低谷期的系统性能，保持系统平稳高效。在实际应用中引入了Kafka消息队列，单台服务器吞吐量高达百万条数据。规则引擎将复杂的水质标准转换为统一格式的引擎文件，转换过程可在实时监控系统之外完成，避免实时数据处理过程中的资源占用，同时可以根据标准变化灵活修改。完成超标判断后，系统将以页面通知的形式向用户展示超标情况，用户也可以查询历史超标记录，这些数据都可以作为用户决策的参考。

水质监测点和各个终端连接到Netty服务器后，会不间断地向服务器发送水质指标数据，这些数据需要向用户展示出来，本系统的Web部分就是用来展示采样点基础数据和监测到的指标数据，采用vue前端框架技术来开发前端页面，结合Websocket技术来动态地展示所有监测实时数据，采样点基础数据和水质指标实时数据以及历史数据分别如图2～图4所示。

本文提供了一个基于Netty的水质实时监控系统的实现方案，针对水质监测点众多、数据上报频率高、实时性及准确性要求高等特点，实现了实时数据传输过程中的高并发，避免了因采集点过多、数据量过大导致的数据处理不及时或系统性能低下等问题，用户可以根据实时数据及超标情况及时做出科学、准确的决策，具有一定实用性。