真正做过环境监测类系统的人都有同感最折腾人的从来不是写几个CRUD页面而是从传感器探头、数据采集终端、通信链路到后端存储、前端渲染这一整条链路里的各种“意外”。最近我一直在打磨一套基于Python的汾河流域水情水质监测系统整体采用Django和Flask的混合架构流域内多个监测站点的水位、流量以及pH、浊度、溶解氧、氨氮等水质指标都需要实时入库、实时上屏、异常预警。核心目标就一句话让数据从现场设备稳定地流到数据库再清晰直观地呈现在用户面前。这套系统适合谁参考准备做物联网监测平台的开发者接了类似环保、水文、农业水利项目但还在纠结架构选型的人以及想搞清楚Django和Flask同时用到底合不合理的同学。我会把我在这个项目里的设计思路、数据模型、采集链路、后端实现、前端可视化和上线后踩过的坑全部摊开来讲。1. 系统整体设计与架构拆解1.1 这类系统到底要解决什么问题流域水情水质监测不是一个“大屏展示”项目它的难点在数据链路。现场设备通常是各种传感器比如压力式水位计、多参数水质探头、流量计它们通过RS485总线或者模拟量输出信号采集终端负责轮询这些传感器再通过4G DTU或者有线网络把数据传到服务器。这个过程中数据格式不统一、设备时钟不准、网络断断续续、传感器偶发漂移都是常见问题。做系统的人如果不把这条链路里的每个环节都想清楚最后必然发现业务功能写完了数据却不可信。平台侧的职责可以分成几块数据接入、数据存储、数据服务、业务管理、预警通知、可视化展示。数据接入要适配不同型号设备的上报协议存储层要保证历史数据能高效查询服务层要给前端和大屏提供稳定的API业务层还要处理站点管理、设备管理、用户权限、预警配置等。说白了这不是一个简单的“网站”而是一个数据系统和业务系统的结合体。1.2 为什么我选Django和Flask混搭很多人一听Django加Flask就头大觉得“一个项目里用两个框架是不是有病”。但我恰恰认为对于这种多角色的系统混搭反而更合理。Flask足够轻启动快路由简单非常适合只做一件事接收采集端上报的实时数据做基本校验后快速落队列。Django则带着完整的ORM、Admin后台、认证授权体系、迁移机制适合承载业务模型和后台管理。实际项目中我没有让两个框架操作同一套数据库连接配置而是让Flask作为“数据接入网关”独立运行只监听内网端口或者直接绑定MQTT主题Django作为“业务中台”提供REST API和后台管理。Flask接收到的数据经过简单校验后写入Redis队列或者直接批量写入MySQLDjango侧的定时任务再做二次处理和预警判断。两个服务之间通过HTTP内部调用联动不需要复杂的分布式框架。选这个方案有个很现实的原因采集层代码往往要跟着现场设备调来调去比如某个型号的传感器指令格式变了、DTU透传模式切换了这些改动如果直接动业务系统风险很大。用Flask把这层隔离出来改采集接入代码不会影响线上业务接口也不会因为导入了一堆Django配置拖慢启动速度。1.3 整体架构与技术栈一览整个系统的技术栈大概是这样的数据采集端传感器 RS485 Modbus RTU协议部分站点用4G DTU透传TCP部分走MQTT网关。接入服务Python Flask paho-mqtt pymodbus Redis。业务后端Python Django Django REST Framework使用JWT做接口认证。数据存储MySQL保存历史监测数据与业务数据Redis做实时缓存与队列。实时推送WebSocket服务前端大屏实时刷新最新监测数据。定时任务APScheduler负责周期扫描、预警判断、数据完整性校验。前端展示ECharts可视化图表Leaflet地图展示站点分布。这里说下选型理由。MySQL完全能扛住这种量级的监测数据一个站点5分钟上报一条综合记录50个站点一天也才14400条完全没有上时序数据库的必要Redis则用来保存“最新值”和“最近N条待处理数据”既能缓解数据库写压力也能让实时查询非常快MQTT做物联网接入协议非常合适自带主题、QoS、断线重连机制比裸TCP好处理得多。2. 数据模型与监测指标怎么设计才不返工2.1 核心数据表结构做这类系统数据模型设计直接决定后面好不好扩展。我在设计阶段把表分成了四组基础信息表、监测数据表、预警记录表、系统权限表。基础信息表包括监测站点表、设备表、指标编码表。监测站点表字段包括站点编码、站点名称、所属河流断面、经度纬度、站点状态。设备表用于挂接传感器和采集终端字段包括设备编码、所属站点、通信方式、协议类型、波特率等参数。指标编码表非常关键它把所有监测指标统一成代码比如pH、DO、NTU、NH3N、WL、FLOW后面所有数据和配置都按指标代码关联。监测数据表是整个系统的核心我设计了两个层级原始采样表和分钟/小时聚合表。原始采样表一条记录包含站点编码、设备编码、指标代码、数值、采集时间、接收时间。聚合表用于统计计算比如小时平均值、日平均值这样后面做趋势分析时不用全量扫描明细表。站点编码和指标代码是关联的灵魂我建议直接用字符串编码可读性强调试时一眼能看出是哪个站的什么指标别用自增ID到处引用否则联调排查问题会非常痛苦。2.2 指标编码与阈值配置指标编码表不只是存个名字它还要携带单位、数据精度、正常范围、预警阈值。我把阈值分成三个级别正常、关注、报警。比如pH正常范围6.5到8.5关注范围6.0到6.5或者8.5到9.0超出这个范围就属于报警。阈值不能写死在代码里因为不同断面、不同季节的限值可能不一样所以我把阈值配置放到指标编码表里并且允许按站点覆写。项目里实际配置的指标大致包括水温、pH、浊度、溶解氧、电导率、氨氮、总磷、水位、流量。每个指标在指标编码表里都有对应字段indicator_code、indicator_name、unit、decimal_places、default_min、default_max、warn_min、warn_max。前后端所有涉及指标的地方都读取这张表新增指标不用改代码改配置就行。2.3 存储选型与查询优化数据量上来以后查询性能主要靠索引。监测数据表我建了组合索引字段顺序是station_id indicator_code measure_time这样既能支撑“某个站点某个指标一段时间内”的查询也能支撑站点大屏的实时曲线。注意不要单独在measure_time上建普通索引单独建会导致组合索引失效的问题不太明显但组合索引对这类查询的收益更大。另一个容易被忽略的点是历史明细表的数据增长。我按照月份做了分表策略表名类似record_data_202501、record_data_202502。应用层写数据时根据月份路由到对应分表查询时先根据时间范围确定表名再执行SQL。这个做法的效果非常明显百万级数据时查询还在秒级以内。如果你不打算引入时序数据库分表是最简单实用的方案。Redis在这里的用法也要说清楚。我用Redis hash保存每个站点每个指标的最新值key形如station:{code}:latestfield是指标代码同时用List保存最近待持久化的数据队列Flask接入数据后先塞进队列再由定时任务批量写入MySQL。这样做的好处是即使数据库短暂抖动数据也不会丢Redis本身有持久化机制现场设备还可以重连补报。3. 采集链路从传感器到服务器的每一步都得盯住3.1 Modbus与4G DTU怎么接现场最常见的接法是多支传感器通过RS485总线挂在同一条总线上采集终端作为Modbus主站轮询每个传感器的寄存器地址。比如某型号pH探头把测量值映射到保持寄存器40001某水位计把水位映射到40010采集终端按从站地址和寄存器地址读取数据。这里有个关键点Modbus RTU基于串口通信读出来的原始值往往需要按比例换算。pH探头返回的可能是放大10倍后的整数水位计返回的可能是毫米值采集脚本里必须维护一份“寄存器地址-指标代码-缩放系数”的映射表统一换算成标准单位后再上报。我一开始就是没仔细看传感器手册以为返回值就是最终值结果有一段pH数据全部需要回炉重算。4G DTU的接入方式也要分情况。支持MQTT的DTU可以直接把串口透传数据包装成MQTT消息推送到服务器只支持TCP透传的设备我就在Flask里开一个TCP server监听DTU的连接按Modbus RTU over TCP的协议解析。无论哪种方式建议给每台DTU配一个固定的clientId或者连接标识方便服务端识别站点。3.2 MQTT订阅与数据清洗Flask接入服务这一层代码量其实不大但对稳定性要求很高。我使用paho-mqtt订阅每个站点的主题主题设计成water/{station_code}/datapayload是JSON字符串包含指标代码、数值、设备时间戳。Flask收到消息后先做基础校验比如站号是否存在、指标代码是否合法、数值范围是否明显异常再计算一个校验时间最后写入Redis队列。例子代码大概长这样import json import paho.mqtt.client as mqtt def on_message(client, userdata, msg): topic msg.topic try: payload json.loads(msg.payload.decode(utf-8)) except Exception: # 解析失败时记录原始报文便于现场回溯 log_warn(decode issue, topictopic, rawmsg.payload[:128]) return station_code topic.split(/)[1] if validate_payload(station_code, payload): redis_client.rpush(data:raw_queue, json.dumps({ station: station_code, indicators: payload[indicators], device_time: payload[ts], server_time: now_str(), })) client mqtt.Client() client.on_message on_message client.connect(broker_host, broker_port, keepalive60) client.subscribe(water//data, qos1) client.loop_forever()这里面的一个经验是原始报文一定要留底。哪怕入库之后发现某个值不正常还能回溯到原始上报数据而不是拿着已处理的数据干瞪眼。我在Redis里放了一个raw_queue同时让日志系统记录异常报文虽然多花了点存储但排查问题效率高了不少。3.3 时间对齐与数据质量治理采集链路最隐蔽的坑是时间。现场设备的时间通常不准甚至有些传感器根本不带时钟只能依靠采集终端打时间戳。如果服务器直接以设备上报时间为准入库后面画曲线时就会出现“凌晨3点突然有个突刺”这种鬼东西其实是设备时钟乱了。我的处理原则是入库统一使用服务器接收时间作为measure_time同时把设备自带时间戳单独存到device_time字段。两个字段在做数据回溯和现场排查时都能用上。如果服务器接收时间和设备时间差的绝对值超过某个阈值比如10分钟就标记该条数据时间戳异常预警和报表默认剔除避免脏数据污染统计结果。数据质量治理还有一层重复上报。DTU误重连、网络超时重发都会导致同一条数据上报两三次。我在写入MySQL时用了唯一约束字段组合是station_id indicator_code server_time重复数据走ON DUPLICATE KEY UPDATE直接覆盖。这个设计简单有效线上跑了很久都没有重复记录。4. Django与Flask混合开发的核心实现4.1 Flask端轻量接入服务的写法Flask服务不加载Django的配置和ORM只用SQLAlchemy连接池或者甚至只用Redis操作。它负责两件事一是接收采集协议上报的数据二是给Django内部提供几个简单的健康检查和指标查询延伸点。为什么要强调“不加载Django配置”因为现场会经常改接入代码如果Flask和Django共用一套复杂配置改一次协议分析就要重启整个业务系统风险太高。我让Flask独立运行在8801端口用systemd守护和Django的8000端口完全隔离。这样现场协议调整时重启Flask在线用户完全无感知。Flask端我顺手加了一个线程池做数据清洗逻辑不重的场景直接在主线程里处理也行但一旦上报频率高比如几秒一条建议把MQTT回调函数的处理逻辑尽量简化只做“解析-校验-入队”三步复杂处理放到后台线程。4.2 Django端业务模型与API设计Django这边负责业务模型、Admin后台和API接口。我用Django REST Framework提供面向web端的接口比如站点列表、实时数据、历史曲线、预警列表、统计报表。JWT认证用来保护非公开接口大屏展示类接口则单独签发只读token不占用用户登录态。核心模型大致是这样class Station(models.Model): code models.CharField(站点编码, max_length32, uniqueTrue) name models.CharField(站点名称, max_length64) longitude models.FloatField(经度) latitude models.FloatField(纬度) section models.CharField(断面位置, max_length128) status models.BooleanField(启用状态, defaultTrue) class Indicator(models.Model): code models.CharField(指标代码, max_length16, uniqueTrue) name models.CharField(指标名称, max_length32) unit models.CharField(单位, max_length16) warn_min models.FloatField(预警下限, nullTrue) warn_max models.FloatField(预警上限, nullTrue) class RecordData(models.Model): station models.ForeignKey(Station, on_deletemodels.CASCADE) indicator models.ForeignKey(Indicator, on_deletemodels.CASCADE) value models.FloatField(数值) measure_time models.DateTimeField(采集时间, db_indexTrue) class Meta: unique_together (station, indicator, measure_time)这个模型设计里站点和指标是外键关联查询接口用DRF的filter参数按站点编码和指标代码过滤。实际海量查询不要走ORM全量查我会在Service层直接写SQL按月份动态拼表名再把结果组装成可视化需要的JSON结构。4.3 权限控制与后台管理的落地Django自带的Admin后台非常有用管理人员可以直接维护站点、指标、阈值配置。但我没把权限开放给所有人而是分了三个角色系统管理员、监测值班员、访客。系统管理员可以改配置和用户监测值班员只能看数据和确认预警访客只看大屏。DRF的权限控制我用了自定义权限类思路是每个API对应一个操作类型然后再判断用户角色是否有对应权限。预警处理接口做了特殊设计值班员确认预警后要填写处理意见系统会记录处理人和处理时间。这个功能在后来的问题追溯里帮了大忙哪条预警是谁确认的、怎么处理的一目了然。后台界面默认的Django Admin样式比较朴素我改了list_display和list_filter让关键字段直接展示再配置了几个自定义Action比如批量启用站点、批量导出监测数据。对内部系统来说这种“够用不花哨”的做法比硬套一个重型前端框架务实得多。5. 可视化大屏与预警联动5.1 用ECharts画实时曲线与站点地图前端可视化我选了ECharts生态成熟、图表类型丰富。大屏页面主要分三块左侧放站点列表和关键指标卡片中间是流域站点地图右侧是选中站点的实时曲线和预警滚动列表。站点地图用Leaflet加OSM瓦片每个站点按经纬度打点颜色代表当前整体状态绿色正常、黄色关注、红色报警。点击地图上的站点标记右侧曲线图会切换成该站点的数据。ECharts的实时曲线我用的是动态追加数据的方式后端WebSocket每推送一条新数据前端就把旧数据移出窗口并追加新点形成滚动效果。这里有个细节曲线图的X轴不要用设备采集时间而要用服务器接收时间。因为设备时间可能跳变一旦跳变前端图表会出现时间倒退或者空隙观感很差。采集时间可以在Tooltip里展示不影响主趋势的连续性。5.2 WebSocket推送与断线重连实时性是大屏的刚需。轮询接口也能做但5秒一次甚至3秒一次对数据库和带宽都不友好。我最终用WebSocket做实时推送。Flask接入服务收到新数据后把数据写一份到Redis发布订阅通道WebSocket服务订阅这个通道再推送给前端。前端使用浏览器原生WebSocket重点要处理断线重连。我用了一个简单方案连接断开后指数退避重连延迟从1秒开始翻倍最多30秒重连成功后立即请求一次最新数据接口把断线期间漏掉的数据补回来。这样现场网络抖动时大屏最多缺几十秒的数据恢复后马上回补。前端订阅逻辑大致是function connectWS() { ws new WebSocket(ws://${location.host}/ws/realtime); ws.onopen () { ws.send(JSON.stringify({ action: subscribe, stations: selectedStations })); }; ws.onclose () { setTimeout(connectWS, Math.min(30, retryDelay) * 1000); retryDelay * 2; }; ws.onmessage (event) { updateRealtimeChart(JSON.parse(event.data)); }; }WebSocket服务本身是独立进程不挂靠在Django或者Flask进程内重启WebSocket服务不会影响数据采集和业务API。部署上用Nginx做WebSocket反向代理配置Upgrade头就行。5.3 预警规则去抖与通知预警如果做得太“灵敏”会变成狼来了。直接拿单条超阈值的数据触发报警经常会出现瞬时波动导致假报警。我的处理方式是同一站点同一指标连续两条数据超阈值才生成一条有效预警而且同一站点同一指标在10分钟内重复触发不重复告警只更新原预警的最后触发时间。预警判断放在APScheduler定时任务里每5分钟跑一次扫描最近10分钟的监测数据用聚合SQL按站点和指标分组找出超标记录。再和已经存在的未处理预警做比对决定是插入新预警还是更新已有预警。这种方式比逐条判断更高效而且杜绝了并行任务同时跑导致重复插入的问题。通知部分用Webhook推送到内部IM机器人消息内容包含站点名称、指标、当前值、阈值、触发时间。消息里一定要带“确认预警”的连接入口值班员点开就能跳转到系统处理页面少一步操作响应就能快一步。6. 部署上线与线上问题排查实录6.1 NginxGunicorn进程守护的部署组合部署方案我用了Nginx反向代理加GunicornDjango服务跑在8000端口Flask接入服务跑在8801端口WebSocket服务独立跑在9000端口。Nginx统一监听80/443按路径把请求转发到对应服务。关键Nginx配置片段upstream django_backend { server 127.0.0.1:8000; } upstream ws_backend { server 127.0.0.1:9000; } server { listen 80; server_name example.internal; location /static { alias /opt/monitor/staticfiles; } location /api/ { proxy_pass http://django_backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } location /ws/ { proxy_pass http://ws_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; } }Gunicorn的配置我用了gevent worker模式worker数量设为CPU核心数的2倍加1超时时间设成60秒。Django接口里如果有一个慢查询卡住不会立刻拖死其他请求。Flask接入服务我直接用了gunicorn加单worker因为它的任务简单多worker反而会导致MQTT回调重复处理同一个消息。进程守护我推荐用systemd而不是简单的nohup。每个服务写一个unit文件设置Restartalways异常退出后自动拉起。服务日志统一输出到journald和日志文件线上问题排查时可以秒级翻日志。6.2 性能调优与索引优化上线初期我遇到过查询变慢的问题集中在监控大屏的“最近24小时曲线”接口。排查下来发现是没走索引SQL执行了全表扫描。加了组合索引之后查询从2秒多降到50毫秒左右这里把索引设计再强调一遍字段顺序一定是station_id indicator_code measure_time查询条件里有没有站点、有没有指标、有没有时间范围都能命中这个索引。数据库连接池也要调优。Django默认的MySQL连接池在并发上来以后容易被打满我改成了用连接池管理并设置最大连接数60队列大小200。同时所有批量写入都走事务每批500条commit一次性能和稳定性都提升不少。还有一个很容易踩的坑Django Admin列表页如果关联到RecordData这类大表加载时会非常慢。解决方案是不要在后台关联查询监测明细表而是单独做一个用只读SQL实现的“数据检索”页面按站点和时间范围查询只返回需要的数据字段。6.3 常见故障与分析速查表我把上线后遇到的典型问题整理成了一张速查表遇到类似情况可以直接对照排查。现象可能原因排查方向处理办法某个站点长时间无数据DTU离线、SIM卡欠费、串口松动查看systemd日志和MQTT主题订阅量远程重启DTU检查现场供电和串口连接数据偶发乱码或数值突变Modbus CRC错误、传感器电源不稳看Flask日志里的原始报文增加CRC校验丢弃传感器端加稳压电源曲线时间轴出现空隙设备时间跳变、网络中断对比server_time和device_time以服务器时间为准断线补发数据预警重复触发去重逻辑没生效查预警表里同一时间点的记录加唯一约束和10分钟去重窗口页面实时数据卡住WebSocket断连未重连看浏览器控制台WebSocket状态启用指数退避重连并补拉最新数据数据库连接数打满连接池太小或慢查询太多查看慢查询日志和连接数监控优化SQL索引扩大连接池上限串口相关的故障排查看起来很低级但实际却是最容易遇到的大问题。有一次某个站点上报的pH值一直在7到9之间乱跳排查了两天最后发现是探头前端被泥沙糊住了。所以线上系统一定要有“长时间趋势异常”的提示让值班员去现场清洗探头单靠程序算法是解决不了物理世界问题的。另一个值得说的经验是监控预警系统自身。我给Flask接入服务加了心跳接口每30秒上报一次服务状态到Django同时内部有一个“数据新鲜度检查”任务如果某个站点超过15分钟没有新数据系统会自动生成一条设备离线告警。这个设计让我不需要时时盯着系统异常发生时第一时间就能收到通知。我个人在实际操作中的体会是流域水情水质监测系统的成败很大程度不在框架选得多先进而在于数据链路是否稳定、数据是否可信、异常是否能被发现。Django和Flask混搭这套组合帮我很好地平衡了接入灵活性和业务开发效率如果你也准备做类似的项目建议先把采集链路和数据模型打磨扎实再考虑花哨的展示功能。最后再分享一个小技巧所有设备的通信参数和校验算法不要只放在代码里一定要同步一份到项目文档中现场人员排查问题时能省下大量沟通成本。