基于Python的Django+Flask混合架构水情水质监测系统设计

真正做过环境监测类系统的人都有同感：最折腾人的从来不是写几个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队列或者直接批量写入MySQL，Django侧的定时任务再做二次处理和预警判断。两个服务之间通过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}:latest，field是指标代码；同时用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}/data，payload是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", topic=topic, raw=msg.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, keepalive=60) client.subscribe("water/+/data", qos=1) 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_length=32, unique=True) name = models.CharField("站点名称", max_length=64) longitude = models.FloatField("经度") latitude = models.FloatField("纬度") section = models.CharField("断面位置", max_length=128) status = models.BooleanField("启用状态", default=True) class Indicator(models.Model): code = models.CharField("指标代码", max_length=16, unique=True) name = models.CharField("指标名称", max_length=32) unit = models.CharField("单位", max_length=16) warn_min = models.FloatField("预警下限", null=True) warn_max = models.FloatField("预警上限", null=True) class RecordData(models.Model): station = models.ForeignKey(Station, on_delete=models.CASCADE) indicator = models.ForeignKey(Indicator, on_delete=models.CASCADE) value = models.FloatField("数值") measure_time = models.DateTimeField("采集时间", db_index=True) 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 Nginx+Gunicorn+进程守护的部署组合

部署方案我用了Nginx反向代理加Gunicorn，Django服务跑在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文件，设置Restart=always，异常退出后自动拉起。服务日志统一输出到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 常见故障与分析速查表

我把上线后遇到的典型问题整理成了一张速查表，遇到类似情况可以直接对照排查。

串口相关的故障排查看起来很低级，但实际却是最容易遇到的大问题。有一次某个站点上报的pH值一直在7到9之间乱跳，排查了两天最后发现是探头前端被泥沙糊住了。所以线上系统一定要有“长时间趋势异常”的提示，让值班员去现场清洗探头，单靠程序算法是解决不了物理世界问题的。

另一个值得说的经验是监控预警系统自身。我给Flask接入服务加了心跳接口，每30秒上报一次服务状态到Django；同时内部有一个“数据新鲜度检查”任务，如果某个站点超过15分钟没有新数据，系统会自动生成一条设备离线告警。这个设计让我不需要时时盯着系统，异常发生时第一时间就能收到通知。

我个人在实际操作中的体会是，流域水情水质监测系统的成败，很大程度不在框架选得多先进，而在于数据链路是否稳定、数据是否可信、异常是否能被发现。Django和Flask混搭这套组合，帮我很好地平衡了接入灵活性和业务开发效率，如果你也准备做类似的项目，建议先把采集链路和数据模型打磨扎实，再考虑花哨的展示功能。最后再分享一个小技巧：所有设备的通信参数和校验算法不要只放在代码里，一定要同步一份到项目文档中，现场人员排查问题时能省下大量沟通成本。