简介面向智慧城市大气环境监测场景的完整Vue.js前端项目将环境数据采集结果通过可视化面板呈现给使用者并围绕空气质量指数、颗粒物浓度、SO2/NOx/O3等指标展示智能预警逻辑适合智慧城市开发学习者、前端数据可视化初学者用于课程设计或毕业设计参考。压缩包内共38个文件以12个Vue组件、5个JavaScript脚本、2个JSON配置为主干辅以HTML/Sass入口页面、README说明、10张PNG图标或示意素材项目包含components、router、utils、assets、public等标准目录整体约7.05MB目录层次合理便于按模块阅读和二次开发。目前已有215人学习下载。通过该项目可以掌握Vue组件化架构、环境监测数据请求封装与图表映射理解AI异常报警与前端事件联动的基本方式即便不直接部署在真实城市中也能利用dashboard页面的布局思路快速搭建具有实时性和交互性的环境监测可视化面板是研究智慧城市可视化交互不错的落地样本。1. 智慧城市大气环境监测系统到底在解决什么问题打开任何一个城市的环境监测中心后台你会发现一个尴尬的事实国控站的数据是准的但国控站太少一个几百万人口的城市可能只有十几个点位它们代表不了每条街道、每个工业园区的真实空气状况。智慧城市大气环境监测系统要解决的不是那把数据测得更准的最后一公里而是把“测得到”变为“连得上、传得回、算得完、看得懂”。它本质上是一套从传感器选型到数据采集、从时序存储到空间插值、从预警规则到可视化大屏的完整链路工程而不是某个单点软件。这个系统的常见形态是一批空气质量微型站也就是网格化监测设备挂在路灯杆、楼顶或者工地围挡上它们内置 PM2.5、PM10、SO₂、NO₂、O₃、CO 传感器和气象五参数模块通过 4G 或者 LoRa 把数据推回平台。平台侧要完成协议解析、数据清洗、超标预警、空间制图最后投到指挥中心大屏和手机端。你拿到一份“智慧城市大气环境监测系统.zip”里面大概率就是这套平台的源码、数据库脚本和部署文档。对做后端和物联网的人来说它真正考验的是数据链路的设计能力和恶劣数据环境下的工程判断力。这篇文章按我自己的实施习惯把这个系统拆成传感器与站点的选型逻辑、从设备到平台的数据链路、存储与分析、可视化大屏与部署验证五个阶段来讲。每一段都会给出可以直接搬走的代码、SQL、配置和参数表也会写清楚哪些地方最容易踩坑。内容不绑定任何特定商业项目但照着做能搭出一套可实际运行的监测系统框架。2. 监测对象、传感器选型与站点布设的工程约束2.1 六项污染物和气象参数先搞清楚你要测什么大气环境监测的国标常规项是六项污染物PM2.5、PM10、SO₂、NO₂、O₃、CO外加气象五参数温度、湿度、风速、风向、气压。在智慧城市场景下还会适当增加 TVOC挥发性有机物总量和噪声因为市民投诉最多的是异味和噪音。理解这些指标的含义比理解传感器更重要因为后面所有阈值设置、数据清洗规则都从这些指标的性质推导出来。每项指标的时间粒度不同PM 浓度在一分钟尺度上就有明显波动而 O₃ 是光化学产物它的日变化曲线通常是单峰午后 14 点到 17 点最高凌晨最低。如果你做的平台预警规则不看时段对 O₃ 全天用同一个阈值那误报率会很难看。SO₂ 主要来自燃煤和工业源CO 来自机动车不完全燃烧选择监测点位时这两者对应的周边环境特征也不同——前者要靠近工业区下风向后者要靠近交通干道。这些特性不是环境科学课上的知识而是你设计数据库标签、配置预警规则时的实际依据。2.2 国控站、微型站、走航设备三类数据源在系统里的角色一个完整的智慧城市大气监测平台数据不是只有一个来源。国控站也就是国家控制空气质量监测站的数据权威、精度高但点位稀疏更新频率通常是小时级微型站是网格化加密的主力成本低、每分钟上报但传感器精度受温湿度和交叉气体干扰走航设备车载或无人机载用于溯源数据是离散路径上的片段。平台侧不能把它们混在一个模型里处理否则会出现“微型站 PM2.5 超标报警国控站却显示良”的矛盾场景。我一般建议的架构是国控站数据作为校正基准和展示层底色微型站数据作为告警和网格分析的主体走航数据单独存只在做污染溯源时叠加到地图上。三者的存储也尽量分开国控站走小时表微型站走分钟级时序表走航数据走轨迹表。这样设计避免了大表查询性能差的问题也符合数据使用频次的差异。2.3 传感器原理选型光散射、电化学、β射线怎么选微型站里最贵的两个传感器是 PM 传感器和气体传感器它们的选型直接决定了数据质量。PM 传感器常见有两种原理光散射法和 β 射线法。光散射法便宜几百到一千元级别、响应快但受相对湿度影响大——高湿度环境下颗粒物吸湿膨胀测量值会虚高β 射线法精度接近国标法但设备体积大、价格高不适合网格化部署。实际项目里微型站用光散射法但需要加装伴热除湿装置并在平台侧用湿度阈值做数据标记。气体传感器常见有电化学和金属氧化物两种。电化学传感器对目标气体选择性好、功耗低是主流选择但有个明显缺点交叉干扰。比如 NO₂ 传感器对 O₃ 也有响应SO₂ 传感器对 NO₂ 有微弱响应。做平台数据清洗时如果不做交叉干扰补偿夏季 O₃ 高发期会出现 NO₂ 数据系统性偏高。解决办法是平台端做温度补偿和交叉干扰修正或者干脆在设备端用四电极传感器从硬件层面抑制——这部分选型时就要确认。2.4 站点布设的密度与标签设计影响下游所有分析微型站布设密度没有统一标准常见做法是城市建成区 1~3 平方公里一个站点工业园区加密到 0.5~1 平方公里。这个密度直接决定后期空间插值的效果站点太稀插值出来的污染分布图就是一团模糊色块站点太密成本扛不住。站点编号和标签设计是平台侧必须提前规划的因为所有展示和告警都要按标签筛选。站点表我通常这样设计这个结构也会在后面建库时直接用到CREATE TABLE station ( station_id VARCHAR(32) PRIMARY KEY, station_name VARCHAR(64) NOT NULL, longitude DECIMAL(10,6) NOT NULL, latitude DECIMAL(10,6) NOT NULL, district VARCHAR(32), station_type TINYINT COMMENT 1-国控 2-市控 3-微型站 4-走航, scenario VARCHAR(16) COMMENT traffic/industrial/residential, install_height DECIMAL(5,1), status TINYINT DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );scenario字段是很多初做项目的人容易忽略的。同样一个 PM2.5 小时均值 80μg/m³在交通站点可能只是晚高峰的正常波动但在居民区就意味着超标事件——告警规则必须能按场景分组配置。install_height也要存因为路边站 2 米高和楼顶站 15 米高测出来的数据差异很大做数据对比分析时这个字段能帮你解释异常。提示站点经纬度不要在应用层用字符串拼接生成要直接存 DECIMAL后端做空间计算距离筛选、插值时效率高很多。MySQL 8.0 以上也支持空间索引但实测下来点位量不超过几千时普通 B 树索引加经纬度范围查询已经够用。3. 从传感器到平台数据采集链路与协议解析3.1 设备端到平台的三种主流链路选型微型站的通信链路选择是采集层第一个决策点。当前项目里主流是三种4G 全网通直连 MQTT、LoRa 网关汇聚后走 4G、以及有线宽带少数室内或园区场景。4G 直连最省事每台设备一张物联网卡平台侧不用管网关和路由LoRa 适合站点密集且分散的工业园区功耗低、穿透力强但需要自建网关和频谱许可运维成本更高。从平台开发角度看我建议统一走 MQTT。原因有三个一是 MQTT 的 QoS 机制能保证数据不丢微型站数据虽然允许少量丢失但分钟级数据一小时丢太多后面做日统计就失真了二是 MQTT 的 Topic 天然适合按站点分组设备上线、离线、数据上报都可以通过不同 Topic 区分三是 EMQX、VerneMQ 等 Broker 的集群方案成熟几十万设备的接入量不需要自己造轮子。如果设备只支持 HTTP 报文推送平台侧加一层协议适配服务做转换但不要指望设备端会自动重传——很多厂商的 HTTP 上报是没有重试机制的。3.2 MQTT 报文格式设计与字段约定设备上报的报文格式往简化核心就是一个 JSON 包。下面是微型站 PM2.5/PM10/气象五参数上报的实际报文格式{ stationId: S010024, deviceTime: 2025-05-18 14:23:05, dataType: AIR_2MIN, payload: { pm25: 42.5, pm10: 68.3, so2: 8.2, no2: 23.1, o3: 96.7, co: 0.68, temp: 26.3, humidity: 48.2, windDir: 156, windSpeed: 2.1, pressure: 1008.3 }, rssi: -67, voltage: 11.8, version: FW_2.3.1 }dataType字段用于区分分钟数据、小时数据还是报警事件。rssi和voltage是设备健康状态信息平台侧要有定时任务扫描这些字段——信号长期低于 -100dBm 或电压低于某阈值说明设备要掉线了运维工单要靠这个触发。很多厂商把设备状态和业务数据混在一起导致业务表被频繁更新拖慢查询建议把设备状态字段拆分到独立表每分钟批量刷新一次。3.3 边缘网关的协议解析与数据预处理脚本如果设备走的是 Modbus RTU 或 DL/T 645 协议工控场景比较常见边缘网关里需要跑一个协议解析程序。常见做法是用 Python 写一个轻量服务从串口或 TCP 读取设备帧解析后重新封装成 MQTT 消息向上推送。这里贴一个 Modbus RTU 读取微型站寄存器并推送 MQTT 的示例核心是为那些没有现成 MQTT 能力的旧设备做适配import minimalmodbus import paho.mqtt.client as mqtt import json import time import struct instrument minimalmodbus.Instrument(/dev/ttyS0, 1) instrument.serial.baudrate 9600 instrument.serial.bytesize 8 instrument.serial.parity N instrument.serial.timeout 1 def read_air_data(): # 从寄存器 0x0100 开始连续读取 10 个寄存器 # 寄存器1-2: PM2.5 原始值(高位低位组合为INT32) # 寄存器3-4: PM10 原始值 raw instrument.read_registers(0x0100, 10, 3) pm25_raw struct.unpack(i, struct.pack(HH, raw[0], raw[1]))[0] pm10_raw struct.unpack(i, struct.pack(HH, raw[2], raw[3]))[0] # 数据分辨率0.1转换为μg/m3 pm25 pm25_raw / 10.0 pm10 pm10_raw / 10.0 return pm25, pm10 client mqtt.Client() client.connect(120.xx.xx.xx, 1883, 60) while True: try: pm25, pm10 read_air_data() payload { stationId: S010024, deviceTime: time.strftime(%Y-%m-%d %H:%M:%S), dataType: AIR_2MIN, payload: {pm25: pm25, pm10: pm10} } client.publish(env/air/S010024/data, json.dumps(payload), qos1) time.sleep(120) except Exception as e: # 串口异常时重连设备同时上报离线状态 print(f[ERROR] {e}) time.sleep(10)代码逻辑说明read_registers一次性把 10 个寄存器读回来比逐个读寄存器快一个数量级。PM2.5 和 PM10 在 Modbus 报文里是 INT32 大端序存储所以用struct.pack重新组合成字节再进行解析。qos1保证消息至少送达一次但要注意如果 Broker 或网络抖动qos1 可能重复投递平台侧消费时要用stationId deviceTime做唯一键去重。提示设备时间用deviceTime而不是平台接收时间serverTime。原因很简单——设备断网重连后会积压一批老数据如果都用接收时间入库统计会乱。平台侧保留两个时间戳业务统计以deviceTime为准。4. 时序数据存储与分析从清洗到空间插值4.1 时序数据库选型为什么不用 MySQL 存原始采样值很多初版代码用 MySQL 存分钟级数据表格一多、数据一涨就出问题。一台微型站每两分钟上报一次一天 720 条1000 个站点一天就是 72 万条。一个月上千万条MySQL 不是存不下而是查询性能会肉眼可见地下滑尤其是做区域聚合查询时全表扫描会直接把数据库拖垮。时序数据库TSDB就是为这个场景设计的常见选择有 InfluxDB、TDengine、TimescaleDB。TDengine 在这个项目里优势更明显Cloud Native 部署简单超级表模型适合描述“同一套采集指标、不同站点”的结构写入吞吐量在单机上就能跑到每秒几万条。而且它对 SQL 的支持比较完整团队切换成本低。InfluxDB 的 Flux 查询语言学习成本相对高如果团队只写过 SQL第一次写 Flux 会很不适应。如果你的公司整体技术栈是 PostgreSQL那 TimescaleDB 是个折中方案——不需要额外维护一套数据库但写入性能比专业 TSDB 还是会弱一些。4.2 TDengine 建表与降采样查询TDengine 的建模逻辑是先建超级表再为每个站点建立子表。子表名用站点 ID这样查询时按站点维度会自动分区性能远高于单张宽表。建表语句如下CREATE DATABASE air_monitor KEEP 365 DURATION 30 BUFFER 16 WAL_LEVEL 2; CREATE STABLE air_quality ( ts TIMESTAMP, device_time TIMESTAMP, pm25 FLOAT, pm10 FLOAT, so2 FLOAT, no2 FLOAT, o3 FLOAT, co FLOAT, temp FLOAT, humidity FLOAT, wind_dir SMALLINT, wind_speed FLOAT, pressure FLOAT ) TAGS (station_id VARCHAR(32), district VARCHAR(16), scenario VARCHAR(16)); CREATE TABLE air_quality_s010024 USING air_quality TAGS (S010024, 朝阳区, traffic); CREATE TABLE air_quality_s010025 USING air_quality TAGS (S010025, 海淀区, residential);注意 TDengine 要求第一列必须是时间戳主键上报时间device_time和入库时间ts要分开存查询时统一按device_time来做时间过滤。查询某个区域的当前小时均值SQL 可以这样写SELECT district, AVG(pm25) AS avg_pm25, MAX(pm25) AS max_pm25, AVG(o3) AS avg_o3 FROM air_quality WHERE device_time NOW() - 1h GROUP BY district;TDengine 的GROUP BY支持按标签分组扫描时只读相关站点的子表性能比 MySQL 按字段 GROUP BY 快很多。但这个查询跑完后得到的是原始值的算术平均——如果要和国标对比还要把均值时段对齐国标 AQI 对应的 PM2.5 是小时均值而小时均值的意思是 60 分钟采样数据的平均所以你的降采样任务要以设备时间为准对齐到自然小时而不是从当前时间往前推 60 分钟。4.3 数据清洗规则异常值剔除与湿度补偿原始采集值不能直接入库参与统计这是数据链路上最容易忽略但影响极大的一环。常见的异常类型有四类传感器掉线产生的零值PM 传感器长时间不出数会输出 0 或负值设备校准时的跳变现场校准会喷标准气体浓度瞬间飙升到几千高湿度导致的 PM 虚高湿度 90% 时光散射法数据基本不能用信号干扰导致的毛刺瞬时值超过正常范围数倍然后又回落清洗规则我一般按顺序执行先过滤物理范围PM 值 0~2000超过范围直接剔除再做一小时窗口滑动中值检测剔除偏离中值 3 倍绝对偏差的数据点最后用湿度阈值标记而非直接删除。直接删除数据会让小时均值失真标记后可以在展示时降权或加备注。下面是一段用 Python 做滑动窗口异常值过滤的示例应用在数据入库前的 Kafka 消费链路里import pandas as pd import numpy as np def clean_series(ts, values, window10min, z_thresh3.0): df pd.DataFrame({ts: ts, val: values}).set_index(ts) # 物理范围剔除 df df[(df[val] 0) (df[val] 2000)] # 滚动中位数 rolling_med df[val].rolling(window, min_periods3, centerTrue).median() rolling_mad (df[val] - rolling_med).abs().rolling(window, min_periods3, centerTrue).median() # 改进的Z-score方法异常值标记为NaN modified_zscore 0.6745 * (df[val] - rolling_med) / (rolling_mad 1e-9) df.loc[modified_zscore.abs() z_thresh, val] np.nan return df逻辑说明rolling window设为中心窗口保证判断每个点时能参考前后各 5 分钟的数据对突变点的识别更稳。1e-9是防止 MAD 为 0 时除零但要注意如果连续多个值完全相同设备卡死MAD 会接近 0这时所有点都会被标记为异常实际效果是把卡死那一段数据全踢掉——对统计来说这是好事。4.4 空间插值用反距离权重画出污染分布热力图站点数据只是离散的要生成一张全市面上的污染分布图必须做空间插值。常用方案有两种反距离权重IDW和克里金插值。IDW 思路简单、计算开销小适合实时大屏——克里金需要拟合变异函数模型计算量大且参数调优依赖人工经验批量跑小时级数据还行实时场景很吃力。微型站的密度一般足够插值误差不是主要矛盾所以我通常选 IDW。下面是用 scikit-learn 实现 IDW 插值并输出网格浓度值的 Python 代码import numpy as np from sklearn.neighbors import NearestNeighbors def idw_interpolate(station_lon, station_lat, station_value, grid_lon, grid_lat, power2.0, k8): xy np.column_stack([station_lon, station_lat]) # 找到每个网格点最近的k个站点 nn NearestNeighbors(n_neighborsk).fit(xy) dist, idx nn.kneighbors(np.column_stack([grid_lon, grid_lat])) weights 1.0 / np.power(dist 1e-8, power) weights / weights.sum(axis1, keepdimsTrue) grid_value np.sum(weights * station_value[idx], axis1) return grid_valuepower是距离衰减幂次默认取 2k取 8 时能平衡局部细节和整体平滑度。有个小坑如果某个网格点距离最近站点只有几米dist接近 0权重会趋向无穷大导致该点颜色异常突兀所以要在距离分母上加一个极小的常数1e-8避免除零。网格分辨率建议取 500 米到 1 公里太细了计算慢且意义不大太粗了看不出街道级别的差异。5. 可视化大屏与预警联动把数据变成可执行的决策动作5.1 大屏组成与前端技术选型智慧城市大气监测大屏的常见构成是底部的全市污染分布热力图由上面板 IDW 插值结果渲染、左侧排名榜站点实时浓度倒序、右侧趋势曲线选定站点近24小时六参数变化、中间顶部放 AQI 等级和首要污染物。展示层我常用 Vue3 加 ECharts 加 Mapbox GL 或 Leaflet地图底图用天地图做背景PM2.5 热力叠加图层用 ECharts GL 的 scatter3D 或 heatmap 渲染。大屏的实时性要求分钟级前端轮询 30 秒一次已经足够——不要用 WebSocket 做纯展示型大屏因为几百个客户端同时订阅一个 topic后端和数据库压力会很大。数据更新频率要和设备上报频率解耦设备两分钟一条大屏展示一分钟刷新一次就没意义。建议后端加一层 Redis 缓存聚合结果大屏接口直接读缓存避免每 30 秒就去查一次 TDengine。5.2 预警规则的引擎化设计预警是监测系统里真正产生业务价值的部分但也是最容易写死的部分。很多项目把预警阈值硬编码在代码里换一个城市或者换一个季节就要改代码发版。成熟做法是把预警规则做成可配置的规则表比如这样CREATE TABLE alert_rule ( rule_id INT PRIMARY KEY AUTO_INCREMENT, rule_name VARCHAR(64), pollutant VARCHAR(16) NOT NULL, scenario VARCHAR(16), threshold_val FLOAT NOT NULL, duration_min INT DEFAULT 0, alert_level TINYINT COMMENT 1-蓝色 2-黄色 3-橙色 4-红色, time_start TIME, time_end TIME, enabled TINYINT DEFAULT 1, updated_at TIMESTAMP );time_start和time_end支持配置特定时段启用的规则应对前面说的 O₃ 午后高发、夜间不预警的场景。有一个容易漏掉的边界是设备离线后重新上线时会补传积压数据这批老数据会瞬间触发多条预警。处理办法是在预警服务里加一个时间窗口过滤只统计当前时间前 10 分钟内的数据超出时间窗口的直接丢弃。预警通知链路完整闭环一般包括预警事件生成、写入告警表、推送到消息队列、分发到短信/企业微信/钉钉、人工确认与归档。告警表里一定要保留rule_id和threshold_val的快照防止规则表后续变更导致历史告警无法回溯。6. 本地部署与数据验证的可用技巧拿到这套系统第一件事不是看代码跑业务而是搭一套最小可运行环境用模拟数据走通全链路。我建议用 Docker Compose 一键编排四个服务EMQXMQTT Broker、TDengine时序库、MySQL业务库、采集服务Python 模拟器。这是一个能稳定运行的最小骨架version: 3.8 services: emqx: image: emqx/emqx:5.8 container_name: emqx ports: - 1883:1883 - 18083:18083 environment: EMQX_DASHBOARD__DEFAULT_PASSWORD: admin123 tdengine: image: tdengine/tdengine:3.3 container_name: tdengine ports: - 6030:6030 - 6041:6041 volumes: - tdengine_data:/var/lib/taos mysql: image: mysql:8.4 container_name: mysql ports: - 3306:3306 environment: MYSQL_ROOT_PASSWORD: root123 MYSQL_DATABASE: air_monitor volumes: - mysql_data:/var/lib/mysql mqtt_simulator: build: ./simulator container_name: mqtt_simulator depends_on: - emqx environment: MQTT_HOST: emqx MQTT_PORT: 1883 STATION_COUNT: 20 restart: unless-stopped volumes: tdengine_data: mysql_data:mqtt_simulator是一个用 Python 写的模拟器容器随机生成 20 个站点的 PM 和气象数据每 5 秒循环推送模拟真实微型站的上报行为。通过这种方式不需要真实硬件就能把整个平台链路跑起来。验证平台数据链路是否正常的检查点有EMQX Dashboardhttp://localhost:18083里能看到 20 个客户端在线消息速率稳定TDengine 里执行SELECT COUNT(*) FROM air_quality WHERE device_time NOW() - 1m返回结果大于 0 且持续增长大屏页面能渲染出热力图且颜色分布与模拟数据的空间分布一致部署完成后的验证顺序建议是先验证消息接入确认 MQTT 消息有进再验证存储TDengine 表有数据再验证清洗和插值输出日志打印最后验证大屏和预警。不要跳过第一步直接查大屏——大屏空白时你没法判断是前端问题、接口问题还是数据根本没有入库。用mosquitto_sub在命令行订阅原始消息能快速定位问题层级mosquitto_sub -h localhost -p 1883 -t env/air/# -v这个命令会打印所有空气监测相关的 MQTT 消息如果这里能看到消息流但 TDengine 无数据问题出在消费入库环节如果这里空无一条问题出在设备端或模拟器。按这个思路排查半小时内能定位百分之八十的链路问题。最后一招TDengine 的taos shell里有个DESCRIBE air_quality命令启动之后先跑一下看看超级表的标签列是否建全——我遇到过不止一次生产环境少建了district标签导致后面按行政区统计的接口全部失效只能重建表补数据。提前确认这一步比上线后补救省力得多。本文还有配套的精品资源点击获取 智慧城市大气环境监测系统全链路搭建：从传感器选型到可视化大屏
简介面向智慧城市大气环境监测场景的完整Vue.js前端项目将环境数据采集结果通过可视化面板呈现给使用者并围绕空气质量指数、颗粒物浓度、SO2/NOx/O3等指标展示智能预警逻辑适合智慧城市开发学习者、前端数据可视化初学者用于课程设计或毕业设计参考。压缩包内共38个文件以12个Vue组件、5个JavaScript脚本、2个JSON配置为主干辅以HTML/Sass入口页面、README说明、10张PNG图标或示意素材项目包含components、router、utils、assets、public等标准目录整体约7.05MB目录层次合理便于按模块阅读和二次开发。目前已有215人学习下载。通过该项目可以掌握Vue组件化架构、环境监测数据请求封装与图表映射理解AI异常报警与前端事件联动的基本方式即便不直接部署在真实城市中也能利用dashboard页面的布局思路快速搭建具有实时性和交互性的环境监测可视化面板是研究智慧城市可视化交互不错的落地样本。1. 智慧城市大气环境监测系统到底在解决什么问题打开任何一个城市的环境监测中心后台你会发现一个尴尬的事实国控站的数据是准的但国控站太少一个几百万人口的城市可能只有十几个点位它们代表不了每条街道、每个工业园区的真实空气状况。智慧城市大气环境监测系统要解决的不是那把数据测得更准的最后一公里而是把“测得到”变为“连得上、传得回、算得完、看得懂”。它本质上是一套从传感器选型到数据采集、从时序存储到空间插值、从预警规则到可视化大屏的完整链路工程而不是某个单点软件。这个系统的常见形态是一批空气质量微型站也就是网格化监测设备挂在路灯杆、楼顶或者工地围挡上它们内置 PM2.5、PM10、SO₂、NO₂、O₃、CO 传感器和气象五参数模块通过 4G 或者 LoRa 把数据推回平台。平台侧要完成协议解析、数据清洗、超标预警、空间制图最后投到指挥中心大屏和手机端。你拿到一份“智慧城市大气环境监测系统.zip”里面大概率就是这套平台的源码、数据库脚本和部署文档。对做后端和物联网的人来说它真正考验的是数据链路的设计能力和恶劣数据环境下的工程判断力。这篇文章按我自己的实施习惯把这个系统拆成传感器与站点的选型逻辑、从设备到平台的数据链路、存储与分析、可视化大屏与部署验证五个阶段来讲。每一段都会给出可以直接搬走的代码、SQL、配置和参数表也会写清楚哪些地方最容易踩坑。内容不绑定任何特定商业项目但照着做能搭出一套可实际运行的监测系统框架。2. 监测对象、传感器选型与站点布设的工程约束2.1 六项污染物和气象参数先搞清楚你要测什么大气环境监测的国标常规项是六项污染物PM2.5、PM10、SO₂、NO₂、O₃、CO外加气象五参数温度、湿度、风速、风向、气压。在智慧城市场景下还会适当增加 TVOC挥发性有机物总量和噪声因为市民投诉最多的是异味和噪音。理解这些指标的含义比理解传感器更重要因为后面所有阈值设置、数据清洗规则都从这些指标的性质推导出来。每项指标的时间粒度不同PM 浓度在一分钟尺度上就有明显波动而 O₃ 是光化学产物它的日变化曲线通常是单峰午后 14 点到 17 点最高凌晨最低。如果你做的平台预警规则不看时段对 O₃ 全天用同一个阈值那误报率会很难看。SO₂ 主要来自燃煤和工业源CO 来自机动车不完全燃烧选择监测点位时这两者对应的周边环境特征也不同——前者要靠近工业区下风向后者要靠近交通干道。这些特性不是环境科学课上的知识而是你设计数据库标签、配置预警规则时的实际依据。2.2 国控站、微型站、走航设备三类数据源在系统里的角色一个完整的智慧城市大气监测平台数据不是只有一个来源。国控站也就是国家控制空气质量监测站的数据权威、精度高但点位稀疏更新频率通常是小时级微型站是网格化加密的主力成本低、每分钟上报但传感器精度受温湿度和交叉气体干扰走航设备车载或无人机载用于溯源数据是离散路径上的片段。平台侧不能把它们混在一个模型里处理否则会出现“微型站 PM2.5 超标报警国控站却显示良”的矛盾场景。我一般建议的架构是国控站数据作为校正基准和展示层底色微型站数据作为告警和网格分析的主体走航数据单独存只在做污染溯源时叠加到地图上。三者的存储也尽量分开国控站走小时表微型站走分钟级时序表走航数据走轨迹表。这样设计避免了大表查询性能差的问题也符合数据使用频次的差异。2.3 传感器原理选型光散射、电化学、β射线怎么选微型站里最贵的两个传感器是 PM 传感器和气体传感器它们的选型直接决定了数据质量。PM 传感器常见有两种原理光散射法和 β 射线法。光散射法便宜几百到一千元级别、响应快但受相对湿度影响大——高湿度环境下颗粒物吸湿膨胀测量值会虚高β 射线法精度接近国标法但设备体积大、价格高不适合网格化部署。实际项目里微型站用光散射法但需要加装伴热除湿装置并在平台侧用湿度阈值做数据标记。气体传感器常见有电化学和金属氧化物两种。电化学传感器对目标气体选择性好、功耗低是主流选择但有个明显缺点交叉干扰。比如 NO₂ 传感器对 O₃ 也有响应SO₂ 传感器对 NO₂ 有微弱响应。做平台数据清洗时如果不做交叉干扰补偿夏季 O₃ 高发期会出现 NO₂ 数据系统性偏高。解决办法是平台端做温度补偿和交叉干扰修正或者干脆在设备端用四电极传感器从硬件层面抑制——这部分选型时就要确认。2.4 站点布设的密度与标签设计影响下游所有分析微型站布设密度没有统一标准常见做法是城市建成区 1~3 平方公里一个站点工业园区加密到 0.5~1 平方公里。这个密度直接决定后期空间插值的效果站点太稀插值出来的污染分布图就是一团模糊色块站点太密成本扛不住。站点编号和标签设计是平台侧必须提前规划的因为所有展示和告警都要按标签筛选。站点表我通常这样设计这个结构也会在后面建库时直接用到CREATE TABLE station ( station_id VARCHAR(32) PRIMARY KEY, station_name VARCHAR(64) NOT NULL, longitude DECIMAL(10,6) NOT NULL, latitude DECIMAL(10,6) NOT NULL, district VARCHAR(32), station_type TINYINT COMMENT 1-国控 2-市控 3-微型站 4-走航, scenario VARCHAR(16) COMMENT traffic/industrial/residential, install_height DECIMAL(5,1), status TINYINT DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );scenario字段是很多初做项目的人容易忽略的。同样一个 PM2.5 小时均值 80μg/m³在交通站点可能只是晚高峰的正常波动但在居民区就意味着超标事件——告警规则必须能按场景分组配置。install_height也要存因为路边站 2 米高和楼顶站 15 米高测出来的数据差异很大做数据对比分析时这个字段能帮你解释异常。提示站点经纬度不要在应用层用字符串拼接生成要直接存 DECIMAL后端做空间计算距离筛选、插值时效率高很多。MySQL 8.0 以上也支持空间索引但实测下来点位量不超过几千时普通 B 树索引加经纬度范围查询已经够用。3. 从传感器到平台数据采集链路与协议解析3.1 设备端到平台的三种主流链路选型微型站的通信链路选择是采集层第一个决策点。当前项目里主流是三种4G 全网通直连 MQTT、LoRa 网关汇聚后走 4G、以及有线宽带少数室内或园区场景。4G 直连最省事每台设备一张物联网卡平台侧不用管网关和路由LoRa 适合站点密集且分散的工业园区功耗低、穿透力强但需要自建网关和频谱许可运维成本更高。从平台开发角度看我建议统一走 MQTT。原因有三个一是 MQTT 的 QoS 机制能保证数据不丢微型站数据虽然允许少量丢失但分钟级数据一小时丢太多后面做日统计就失真了二是 MQTT 的 Topic 天然适合按站点分组设备上线、离线、数据上报都可以通过不同 Topic 区分三是 EMQX、VerneMQ 等 Broker 的集群方案成熟几十万设备的接入量不需要自己造轮子。如果设备只支持 HTTP 报文推送平台侧加一层协议适配服务做转换但不要指望设备端会自动重传——很多厂商的 HTTP 上报是没有重试机制的。3.2 MQTT 报文格式设计与字段约定设备上报的报文格式往简化核心就是一个 JSON 包。下面是微型站 PM2.5/PM10/气象五参数上报的实际报文格式{ stationId: S010024, deviceTime: 2025-05-18 14:23:05, dataType: AIR_2MIN, payload: { pm25: 42.5, pm10: 68.3, so2: 8.2, no2: 23.1, o3: 96.7, co: 0.68, temp: 26.3, humidity: 48.2, windDir: 156, windSpeed: 2.1, pressure: 1008.3 }, rssi: -67, voltage: 11.8, version: FW_2.3.1 }dataType字段用于区分分钟数据、小时数据还是报警事件。rssi和voltage是设备健康状态信息平台侧要有定时任务扫描这些字段——信号长期低于 -100dBm 或电压低于某阈值说明设备要掉线了运维工单要靠这个触发。很多厂商把设备状态和业务数据混在一起导致业务表被频繁更新拖慢查询建议把设备状态字段拆分到独立表每分钟批量刷新一次。3.3 边缘网关的协议解析与数据预处理脚本如果设备走的是 Modbus RTU 或 DL/T 645 协议工控场景比较常见边缘网关里需要跑一个协议解析程序。常见做法是用 Python 写一个轻量服务从串口或 TCP 读取设备帧解析后重新封装成 MQTT 消息向上推送。这里贴一个 Modbus RTU 读取微型站寄存器并推送 MQTT 的示例核心是为那些没有现成 MQTT 能力的旧设备做适配import minimalmodbus import paho.mqtt.client as mqtt import json import time import struct instrument minimalmodbus.Instrument(/dev/ttyS0, 1) instrument.serial.baudrate 9600 instrument.serial.bytesize 8 instrument.serial.parity N instrument.serial.timeout 1 def read_air_data(): # 从寄存器 0x0100 开始连续读取 10 个寄存器 # 寄存器1-2: PM2.5 原始值(高位低位组合为INT32) # 寄存器3-4: PM10 原始值 raw instrument.read_registers(0x0100, 10, 3) pm25_raw struct.unpack(i, struct.pack(HH, raw[0], raw[1]))[0] pm10_raw struct.unpack(i, struct.pack(HH, raw[2], raw[3]))[0] # 数据分辨率0.1转换为μg/m3 pm25 pm25_raw / 10.0 pm10 pm10_raw / 10.0 return pm25, pm10 client mqtt.Client() client.connect(120.xx.xx.xx, 1883, 60) while True: try: pm25, pm10 read_air_data() payload { stationId: S010024, deviceTime: time.strftime(%Y-%m-%d %H:%M:%S), dataType: AIR_2MIN, payload: {pm25: pm25, pm10: pm10} } client.publish(env/air/S010024/data, json.dumps(payload), qos1) time.sleep(120) except Exception as e: # 串口异常时重连设备同时上报离线状态 print(f[ERROR] {e}) time.sleep(10)代码逻辑说明read_registers一次性把 10 个寄存器读回来比逐个读寄存器快一个数量级。PM2.5 和 PM10 在 Modbus 报文里是 INT32 大端序存储所以用struct.pack重新组合成字节再进行解析。qos1保证消息至少送达一次但要注意如果 Broker 或网络抖动qos1 可能重复投递平台侧消费时要用stationId deviceTime做唯一键去重。提示设备时间用deviceTime而不是平台接收时间serverTime。原因很简单——设备断网重连后会积压一批老数据如果都用接收时间入库统计会乱。平台侧保留两个时间戳业务统计以deviceTime为准。4. 时序数据存储与分析从清洗到空间插值4.1 时序数据库选型为什么不用 MySQL 存原始采样值很多初版代码用 MySQL 存分钟级数据表格一多、数据一涨就出问题。一台微型站每两分钟上报一次一天 720 条1000 个站点一天就是 72 万条。一个月上千万条MySQL 不是存不下而是查询性能会肉眼可见地下滑尤其是做区域聚合查询时全表扫描会直接把数据库拖垮。时序数据库TSDB就是为这个场景设计的常见选择有 InfluxDB、TDengine、TimescaleDB。TDengine 在这个项目里优势更明显Cloud Native 部署简单超级表模型适合描述“同一套采集指标、不同站点”的结构写入吞吐量在单机上就能跑到每秒几万条。而且它对 SQL 的支持比较完整团队切换成本低。InfluxDB 的 Flux 查询语言学习成本相对高如果团队只写过 SQL第一次写 Flux 会很不适应。如果你的公司整体技术栈是 PostgreSQL那 TimescaleDB 是个折中方案——不需要额外维护一套数据库但写入性能比专业 TSDB 还是会弱一些。4.2 TDengine 建表与降采样查询TDengine 的建模逻辑是先建超级表再为每个站点建立子表。子表名用站点 ID这样查询时按站点维度会自动分区性能远高于单张宽表。建表语句如下CREATE DATABASE air_monitor KEEP 365 DURATION 30 BUFFER 16 WAL_LEVEL 2; CREATE STABLE air_quality ( ts TIMESTAMP, device_time TIMESTAMP, pm25 FLOAT, pm10 FLOAT, so2 FLOAT, no2 FLOAT, o3 FLOAT, co FLOAT, temp FLOAT, humidity FLOAT, wind_dir SMALLINT, wind_speed FLOAT, pressure FLOAT ) TAGS (station_id VARCHAR(32), district VARCHAR(16), scenario VARCHAR(16)); CREATE TABLE air_quality_s010024 USING air_quality TAGS (S010024, 朝阳区, traffic); CREATE TABLE air_quality_s010025 USING air_quality TAGS (S010025, 海淀区, residential);注意 TDengine 要求第一列必须是时间戳主键上报时间device_time和入库时间ts要分开存查询时统一按device_time来做时间过滤。查询某个区域的当前小时均值SQL 可以这样写SELECT district, AVG(pm25) AS avg_pm25, MAX(pm25) AS max_pm25, AVG(o3) AS avg_o3 FROM air_quality WHERE device_time NOW() - 1h GROUP BY district;TDengine 的GROUP BY支持按标签分组扫描时只读相关站点的子表性能比 MySQL 按字段 GROUP BY 快很多。但这个查询跑完后得到的是原始值的算术平均——如果要和国标对比还要把均值时段对齐国标 AQI 对应的 PM2.5 是小时均值而小时均值的意思是 60 分钟采样数据的平均所以你的降采样任务要以设备时间为准对齐到自然小时而不是从当前时间往前推 60 分钟。4.3 数据清洗规则异常值剔除与湿度补偿原始采集值不能直接入库参与统计这是数据链路上最容易忽略但影响极大的一环。常见的异常类型有四类传感器掉线产生的零值PM 传感器长时间不出数会输出 0 或负值设备校准时的跳变现场校准会喷标准气体浓度瞬间飙升到几千高湿度导致的 PM 虚高湿度 90% 时光散射法数据基本不能用信号干扰导致的毛刺瞬时值超过正常范围数倍然后又回落清洗规则我一般按顺序执行先过滤物理范围PM 值 0~2000超过范围直接剔除再做一小时窗口滑动中值检测剔除偏离中值 3 倍绝对偏差的数据点最后用湿度阈值标记而非直接删除。直接删除数据会让小时均值失真标记后可以在展示时降权或加备注。下面是一段用 Python 做滑动窗口异常值过滤的示例应用在数据入库前的 Kafka 消费链路里import pandas as pd import numpy as np def clean_series(ts, values, window10min, z_thresh3.0): df pd.DataFrame({ts: ts, val: values}).set_index(ts) # 物理范围剔除 df df[(df[val] 0) (df[val] 2000)] # 滚动中位数 rolling_med df[val].rolling(window, min_periods3, centerTrue).median() rolling_mad (df[val] - rolling_med).abs().rolling(window, min_periods3, centerTrue).median() # 改进的Z-score方法异常值标记为NaN modified_zscore 0.6745 * (df[val] - rolling_med) / (rolling_mad 1e-9) df.loc[modified_zscore.abs() z_thresh, val] np.nan return df逻辑说明rolling window设为中心窗口保证判断每个点时能参考前后各 5 分钟的数据对突变点的识别更稳。1e-9是防止 MAD 为 0 时除零但要注意如果连续多个值完全相同设备卡死MAD 会接近 0这时所有点都会被标记为异常实际效果是把卡死那一段数据全踢掉——对统计来说这是好事。4.4 空间插值用反距离权重画出污染分布热力图站点数据只是离散的要生成一张全市面上的污染分布图必须做空间插值。常用方案有两种反距离权重IDW和克里金插值。IDW 思路简单、计算开销小适合实时大屏——克里金需要拟合变异函数模型计算量大且参数调优依赖人工经验批量跑小时级数据还行实时场景很吃力。微型站的密度一般足够插值误差不是主要矛盾所以我通常选 IDW。下面是用 scikit-learn 实现 IDW 插值并输出网格浓度值的 Python 代码import numpy as np from sklearn.neighbors import NearestNeighbors def idw_interpolate(station_lon, station_lat, station_value, grid_lon, grid_lat, power2.0, k8): xy np.column_stack([station_lon, station_lat]) # 找到每个网格点最近的k个站点 nn NearestNeighbors(n_neighborsk).fit(xy) dist, idx nn.kneighbors(np.column_stack([grid_lon, grid_lat])) weights 1.0 / np.power(dist 1e-8, power) weights / weights.sum(axis1, keepdimsTrue) grid_value np.sum(weights * station_value[idx], axis1) return grid_valuepower是距离衰减幂次默认取 2k取 8 时能平衡局部细节和整体平滑度。有个小坑如果某个网格点距离最近站点只有几米dist接近 0权重会趋向无穷大导致该点颜色异常突兀所以要在距离分母上加一个极小的常数1e-8避免除零。网格分辨率建议取 500 米到 1 公里太细了计算慢且意义不大太粗了看不出街道级别的差异。5. 可视化大屏与预警联动把数据变成可执行的决策动作5.1 大屏组成与前端技术选型智慧城市大气监测大屏的常见构成是底部的全市污染分布热力图由上面板 IDW 插值结果渲染、左侧排名榜站点实时浓度倒序、右侧趋势曲线选定站点近24小时六参数变化、中间顶部放 AQI 等级和首要污染物。展示层我常用 Vue3 加 ECharts 加 Mapbox GL 或 Leaflet地图底图用天地图做背景PM2.5 热力叠加图层用 ECharts GL 的 scatter3D 或 heatmap 渲染。大屏的实时性要求分钟级前端轮询 30 秒一次已经足够——不要用 WebSocket 做纯展示型大屏因为几百个客户端同时订阅一个 topic后端和数据库压力会很大。数据更新频率要和设备上报频率解耦设备两分钟一条大屏展示一分钟刷新一次就没意义。建议后端加一层 Redis 缓存聚合结果大屏接口直接读缓存避免每 30 秒就去查一次 TDengine。5.2 预警规则的引擎化设计预警是监测系统里真正产生业务价值的部分但也是最容易写死的部分。很多项目把预警阈值硬编码在代码里换一个城市或者换一个季节就要改代码发版。成熟做法是把预警规则做成可配置的规则表比如这样CREATE TABLE alert_rule ( rule_id INT PRIMARY KEY AUTO_INCREMENT, rule_name VARCHAR(64), pollutant VARCHAR(16) NOT NULL, scenario VARCHAR(16), threshold_val FLOAT NOT NULL, duration_min INT DEFAULT 0, alert_level TINYINT COMMENT 1-蓝色 2-黄色 3-橙色 4-红色, time_start TIME, time_end TIME, enabled TINYINT DEFAULT 1, updated_at TIMESTAMP );time_start和time_end支持配置特定时段启用的规则应对前面说的 O₃ 午后高发、夜间不预警的场景。有一个容易漏掉的边界是设备离线后重新上线时会补传积压数据这批老数据会瞬间触发多条预警。处理办法是在预警服务里加一个时间窗口过滤只统计当前时间前 10 分钟内的数据超出时间窗口的直接丢弃。预警通知链路完整闭环一般包括预警事件生成、写入告警表、推送到消息队列、分发到短信/企业微信/钉钉、人工确认与归档。告警表里一定要保留rule_id和threshold_val的快照防止规则表后续变更导致历史告警无法回溯。6. 本地部署与数据验证的可用技巧拿到这套系统第一件事不是看代码跑业务而是搭一套最小可运行环境用模拟数据走通全链路。我建议用 Docker Compose 一键编排四个服务EMQXMQTT Broker、TDengine时序库、MySQL业务库、采集服务Python 模拟器。这是一个能稳定运行的最小骨架version: 3.8 services: emqx: image: emqx/emqx:5.8 container_name: emqx ports: - 1883:1883 - 18083:18083 environment: EMQX_DASHBOARD__DEFAULT_PASSWORD: admin123 tdengine: image: tdengine/tdengine:3.3 container_name: tdengine ports: - 6030:6030 - 6041:6041 volumes: - tdengine_data:/var/lib/taos mysql: image: mysql:8.4 container_name: mysql ports: - 3306:3306 environment: MYSQL_ROOT_PASSWORD: root123 MYSQL_DATABASE: air_monitor volumes: - mysql_data:/var/lib/mysql mqtt_simulator: build: ./simulator container_name: mqtt_simulator depends_on: - emqx environment: MQTT_HOST: emqx MQTT_PORT: 1883 STATION_COUNT: 20 restart: unless-stopped volumes: tdengine_data: mysql_data:mqtt_simulator是一个用 Python 写的模拟器容器随机生成 20 个站点的 PM 和气象数据每 5 秒循环推送模拟真实微型站的上报行为。通过这种方式不需要真实硬件就能把整个平台链路跑起来。验证平台数据链路是否正常的检查点有EMQX Dashboardhttp://localhost:18083里能看到 20 个客户端在线消息速率稳定TDengine 里执行SELECT COUNT(*) FROM air_quality WHERE device_time NOW() - 1m返回结果大于 0 且持续增长大屏页面能渲染出热力图且颜色分布与模拟数据的空间分布一致部署完成后的验证顺序建议是先验证消息接入确认 MQTT 消息有进再验证存储TDengine 表有数据再验证清洗和插值输出日志打印最后验证大屏和预警。不要跳过第一步直接查大屏——大屏空白时你没法判断是前端问题、接口问题还是数据根本没有入库。用mosquitto_sub在命令行订阅原始消息能快速定位问题层级mosquitto_sub -h localhost -p 1883 -t env/air/# -v这个命令会打印所有空气监测相关的 MQTT 消息如果这里能看到消息流但 TDengine 无数据问题出在消费入库环节如果这里空无一条问题出在设备端或模拟器。按这个思路排查半小时内能定位百分之八十的链路问题。最后一招TDengine 的taos shell里有个DESCRIBE air_quality命令启动之后先跑一下看看超级表的标签列是否建全——我遇到过不止一次生产环境少建了district标签导致后面按行政区统计的接口全部失效只能重建表补数据。提前确认这一步比上线后补救省力得多。本文还有配套的精品资源点击获取