真正做過環(huán)境監(jiān)測類系統(tǒng)的人都有同感最折騰人的從來不是寫幾個CRUD頁面而是從傳感器探頭、數(shù)據(jù)采集終端、通信鏈路到后端存儲、前端渲染這一整條鏈路里的各種“意外”。最近我一直在打磨一套基于Python的汾河流域水情水質(zhì)監(jiān)測系統(tǒng)整體采用Django和Flask的混合架構(gòu)流域內(nèi)多個監(jiān)測站點(diǎn)的水位、流量以及pH、濁度、溶解氧、氨氮等水質(zhì)指標(biāo)都需要實(shí)時入庫、實(shí)時上屏、異常預(yù)警。核心目標(biāo)就一句話讓數(shù)據(jù)從現(xiàn)場設(shè)備穩(wěn)定地流到數(shù)據(jù)庫再清晰直觀地呈現(xiàn)在用戶面前。這套系統(tǒng)適合誰參考準(zhǔn)備做物聯(lián)網(wǎng)監(jiān)測平臺的開發(fā)者接了類似環(huán)保、水文、農(nóng)業(yè)水利項(xiàng)目但還在糾結(jié)架構(gòu)選型的人以及想搞清楚Django和Flask同時用到底合不合理的同學(xué)。我會把我在這個項(xiàng)目里的設(shè)計(jì)思路、數(shù)據(jù)模型、采集鏈路、后端實(shí)現(xiàn)、前端可視化和上線后踩過的坑全部攤開來講。1. 系統(tǒng)整體設(shè)計(jì)與架構(gòu)拆解1.1 這類系統(tǒng)到底要解決什么問題流域水情水質(zhì)監(jiān)測不是一個“大屏展示”項(xiàng)目它的難點(diǎn)在數(shù)據(jù)鏈路?，F(xiàn)場設(shè)備通常是各種傳感器比如壓力式水位計(jì)、多參數(shù)水質(zhì)探頭、流量計(jì)它們通過RS485總線或者模擬量輸出信號采集終端負(fù)責(zé)輪詢這些傳感器再通過4G DTU或者有線網(wǎng)絡(luò)把數(shù)據(jù)傳到服務(wù)器。這個過程中數(shù)據(jù)格式不統(tǒng)一、設(shè)備時鐘不準(zhǔn)、網(wǎng)絡(luò)斷斷續(xù)續(xù)、傳感器偶發(fā)漂移都是常見問題。做系統(tǒng)的人如果不把這條鏈路里的每個環(huán)節(jié)都想清楚最后必然發(fā)現(xiàn)業(yè)務(wù)功能寫完了數(shù)據(jù)卻不可信。平臺側(cè)的職責(zé)可以分成幾塊數(shù)據(jù)接入、數(shù)據(jù)存儲、數(shù)據(jù)服務(wù)、業(yè)務(wù)管理、預(yù)警通知、可視化展示。數(shù)據(jù)接入要適配不同型號設(shè)備的上報(bào)協(xié)議存儲層要保證歷史數(shù)據(jù)能高效查詢服務(wù)層要給前端和大屏提供穩(wěn)定的API業(yè)務(wù)層還要處理站點(diǎn)管理、設(shè)備管理、用戶權(quán)限、預(yù)警配置等。說白了這不是一個簡單的“網(wǎng)站”而是一個數(shù)據(jù)系統(tǒng)和業(yè)務(wù)系統(tǒng)的結(jié)合體。1.2 為什么我選Django和Flask混搭很多人一聽Django加Flask就頭大覺得“一個項(xiàng)目里用兩個框架是不是有病”。但我恰恰認(rèn)為對于這種多角色的系統(tǒng)混搭反而更合理。Flask足夠輕啟動快路由簡單非常適合只做一件事接收采集端上報(bào)的實(shí)時數(shù)據(jù)做基本校驗(yàn)后快速落隊(duì)列。Django則帶著完整的ORM、Admin后臺、認(rèn)證授權(quán)體系、遷移機(jī)制適合承載業(yè)務(wù)模型和后臺管理。實(shí)際項(xiàng)目中我沒有讓兩個框架操作同一套數(shù)據(jù)庫連接配置而是讓Flask作為“數(shù)據(jù)接入網(wǎng)關(guān)”獨(dú)立運(yùn)行只監(jiān)聽內(nèi)網(wǎng)端口或者直接綁定MQTT主題Django作為“業(yè)務(wù)中臺”提供REST API和后臺管理。Flask接收到的數(shù)據(jù)經(jīng)過簡單校驗(yàn)后寫入Redis隊(duì)列或者直接批量寫入MySQLDjango側(cè)的定時任務(wù)再做二次處理和預(yù)警判斷。兩個服務(wù)之間通過HTTP內(nèi)部調(diào)用聯(lián)動不需要復(fù)雜的分布式框架。選這個方案有個很現(xiàn)實(shí)的原因采集層代碼往往要跟著現(xiàn)場設(shè)備調(diào)來調(diào)去比如某個型號的傳感器指令格式變了、DTU透傳模式切換了這些改動如果直接動業(yè)務(wù)系統(tǒng)風(fēng)險很大。用Flask把這層隔離出來改采集接入代碼不會影響線上業(yè)務(wù)接口也不會因?yàn)閷?dǎo)入了一堆Django配置拖慢啟動速度。1.3 整體架構(gòu)與技術(shù)棧一覽整個系統(tǒng)的技術(shù)棧大概是這樣的數(shù)據(jù)采集端傳感器 RS485 Modbus RTU協(xié)議部分站點(diǎn)用4G DTU透傳TCP部分走M(jìn)QTT網(wǎng)關(guān)。接入服務(wù)Python Flask paho-mqtt pymodbus Redis。業(yè)務(wù)后端Python Django Django REST Framework使用JWT做接口認(rèn)證。數(shù)據(jù)存儲MySQL保存歷史監(jiān)測數(shù)據(jù)與業(yè)務(wù)數(shù)據(jù)Redis做實(shí)時緩存與隊(duì)列。實(shí)時推送WebSocket服務(wù)前端大屏實(shí)時刷新最新監(jiān)測數(shù)據(jù)。定時任務(wù)APScheduler負(fù)責(zé)周期掃描、預(yù)警判斷、數(shù)據(jù)完整性校驗(yàn)。前端展示ECharts可視化圖表Leaflet地圖展示站點(diǎn)分布。這里說下選型理由。MySQL完全能扛住這種量級的監(jiān)測數(shù)據(jù)一個站點(diǎn)5分鐘上報(bào)一條綜合記錄50個站點(diǎn)一天也才14400條完全沒有上時序數(shù)據(jù)庫的必要Redis則用來保存“最新值”和“最近N條待處理數(shù)據(jù)”既能緩解數(shù)據(jù)庫寫壓力也能讓實(shí)時查詢非?？霱QTT做物聯(lián)網(wǎng)接入?yún)f(xié)議非常合適自帶主題、QoS、斷線重連機(jī)制比裸TCP好處理得多。2. 數(shù)據(jù)模型與監(jiān)測指標(biāo)怎么設(shè)計(jì)才不返工2.1 核心數(shù)據(jù)表結(jié)構(gòu)做這類系統(tǒng)數(shù)據(jù)模型設(shè)計(jì)直接決定后面好不好擴(kuò)展。我在設(shè)計(jì)階段把表分成了四組基礎(chǔ)信息表、監(jiān)測數(shù)據(jù)表、預(yù)警記錄表、系統(tǒng)權(quán)限表?；A(chǔ)信息表包括監(jiān)測站點(diǎn)表、設(shè)備表、指標(biāo)編碼表。監(jiān)測站點(diǎn)表字段包括站點(diǎn)編碼、站點(diǎn)名稱、所屬河流斷面、經(jīng)度緯度、站點(diǎn)狀態(tài)。設(shè)備表用于掛接傳感器和采集終端字段包括設(shè)備編碼、所屬站點(diǎn)、通信方式、協(xié)議類型、波特率等參數(shù)。指標(biāo)編碼表非常關(guān)鍵它把所有監(jiān)測指標(biāo)統(tǒng)一成代碼比如pH、DO、NTU、NH3N、WL、FLOW后面所有數(shù)據(jù)和配置都按指標(biāo)代碼關(guān)聯(lián)。監(jiān)測數(shù)據(jù)表是整個系統(tǒng)的核心我設(shè)計(jì)了兩個層級原始采樣表和分鐘/小時聚合表。原始采樣表一條記錄包含站點(diǎn)編碼、設(shè)備編碼、指標(biāo)代碼、數(shù)值、采集時間、接收時間。聚合表用于統(tǒng)計(jì)計(jì)算比如小時平均值、日平均值這樣后面做趨勢分析時不用全量掃描明細(xì)表。站點(diǎn)編碼和指標(biāo)代碼是關(guān)聯(lián)的靈魂我建議直接用字符串編碼可讀性強(qiáng)調(diào)試時一眼能看出是哪個站的什么指標(biāo)別用自增ID到處引用否則聯(lián)調(diào)排查問題會非常痛苦。2.2 指標(biāo)編碼與閾值配置指標(biāo)編碼表不只是存?zhèn)€名字它還要攜帶單位、數(shù)據(jù)精度、正常范圍、預(yù)警閾值。我把閾值分成三個級別正常、關(guān)注、報(bào)警。比如pH正常范圍6.5到8.5關(guān)注范圍6.0到6.5或者8.5到9.0超出這個范圍就屬于報(bào)警。閾值不能寫死在代碼里因?yàn)椴煌瑪嗝妗⒉煌竟?jié)的限值可能不一樣所以我把閾值配置放到指標(biāo)編碼表里并且允許按站點(diǎn)覆寫。項(xiàng)目里實(shí)際配置的指標(biāo)大致包括水溫、pH、濁度、溶解氧、電導(dǎo)率、氨氮、總磷、水位、流量。每個指標(biāo)在指標(biāo)編碼表里都有對應(yīng)字段indicator_code、indicator_name、unit、decimal_places、default_min、default_max、warn_min、warn_max。前后端所有涉及指標(biāo)的地方都讀取這張表新增指標(biāo)不用改代碼改配置就行。2.3 存儲選型與查詢優(yōu)化數(shù)據(jù)量上來以后查詢性能主要靠索引。監(jiān)測數(shù)據(jù)表我建了組合索引字段順序是station_id indicator_code measure_time這樣既能支撐“某個站點(diǎn)某個指標(biāo)一段時間內(nèi)”的查詢也能支撐站點(diǎn)大屏的實(shí)時曲線。注意不要單獨(dú)在measure_time上建普通索引單獨(dú)建會導(dǎo)致組合索引失效的問題不太明顯但組合索引對這類查詢的收益更大。另一個容易被忽略的點(diǎn)是歷史明細(xì)表的數(shù)據(jù)增長。我按照月份做了分表策略表名類似record_data_202501、record_data_202502。應(yīng)用層寫數(shù)據(jù)時根據(jù)月份路由到對應(yīng)分表查詢時先根據(jù)時間范圍確定表名再執(zhí)行SQL。這個做法的效果非常明顯百萬級數(shù)據(jù)時查詢還在秒級以內(nèi)。如果你不打算引入時序數(shù)據(jù)庫分表是最簡單實(shí)用的方案。Redis在這里的用法也要說清楚。我用Redis hash保存每個站點(diǎn)每個指標(biāo)的最新值key形如station:{code}:latestfield是指標(biāo)代碼同時用List保存最近待持久化的數(shù)據(jù)隊(duì)列Flask接入數(shù)據(jù)后先塞進(jìn)隊(duì)列再由定時任務(wù)批量寫入MySQL。這樣做的好處是即使數(shù)據(jù)庫短暫抖動數(shù)據(jù)也不會丟Redis本身有持久化機(jī)制現(xiàn)場設(shè)備還可以重連補(bǔ)報(bào)。3. 采集鏈路從傳感器到服務(wù)器的每一步都得盯住3.1 Modbus與4G DTU怎么接現(xiàn)場最常見的接法是多支傳感器通過RS485總線掛在同一條總線上采集終端作為Modbus主站輪詢每個傳感器的寄存器地址。比如某型號pH探頭把測量值映射到保持寄存器40001某水位計(jì)把水位映射到40010采集終端按從站地址和寄存器地址讀取數(shù)據(jù)。這里有個關(guān)鍵點(diǎn)Modbus RTU基于串口通信讀出來的原始值往往需要按比例換算。pH探頭返回的可能是放大10倍后的整數(shù)水位計(jì)返回的可能是毫米值采集腳本里必須維護(hù)一份“寄存器地址-指標(biāo)代碼-縮放系數(shù)”的映射表統(tǒng)一換算成標(biāo)準(zhǔn)單位后再上報(bào)。我一開始就是沒仔細(xì)看傳感器手冊以為返回值就是最終值結(jié)果有一段pH數(shù)據(jù)全部需要回爐重算。4G DTU的接入方式也要分情況。支持MQTT的DTU可以直接把串口透傳數(shù)據(jù)包裝成MQTT消息推送到服務(wù)器只支持TCP透傳的設(shè)備我就在Flask里開一個TCP server監(jiān)聽DTU的連接按Modbus RTU over TCP的協(xié)議解析。無論哪種方式建議給每臺DTU配一個固定的clientId或者連接標(biāo)識方便服務(wù)端識別站點(diǎn)。3.2 MQTT訂閱與數(shù)據(jù)清洗Flask接入服務(wù)這一層代碼量其實(shí)不大但對穩(wěn)定性要求很高。我使用paho-mqtt訂閱每個站點(diǎn)的主題主題設(shè)計(jì)成water/{station_code}/datapayload是JSON字符串包含指標(biāo)代碼、數(shù)值、設(shè)備時間戳。Flask收到消息后先做基礎(chǔ)校驗(yàn)比如站號是否存在、指標(biāo)代碼是否合法、數(shù)值范圍是否明顯異常再計(jì)算一個校驗(yàn)時間最后寫入Redis隊(duì)列。例子代碼大概長這樣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: # 解析失敗時記錄原始報(bào)文便于現(xiàn)場回溯 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()這里面的一個經(jīng)驗(yàn)是原始報(bào)文一定要留底。哪怕入庫之后發(fā)現(xiàn)某個值不正常還能回溯到原始上報(bào)數(shù)據(jù)而不是拿著已處理的數(shù)據(jù)干瞪眼。我在Redis里放了一個raw_queue同時讓日志系統(tǒng)記錄異常報(bào)文雖然多花了點(diǎn)存儲但排查問題效率高了不少。3.3 時間對齊與數(shù)據(jù)質(zhì)量治理采集鏈路最隱蔽的坑是時間。現(xiàn)場設(shè)備的時間通常不準(zhǔn)甚至有些傳感器根本不帶時鐘只能依靠采集終端打時間戳。如果服務(wù)器直接以設(shè)備上報(bào)時間為準(zhǔn)入庫后面畫曲線時就會出現(xiàn)“凌晨3點(diǎn)突然有個突刺”這種鬼東西其實(shí)是設(shè)備時鐘亂了。我的處理原則是入庫統(tǒng)一使用服務(wù)器接收時間作為measure_time同時把設(shè)備自帶時間戳單獨(dú)存到device_time字段。兩個字段在做數(shù)據(jù)回溯和現(xiàn)場排查時都能用上。如果服務(wù)器接收時間和設(shè)備時間差的絕對值超過某個閾值比如10分鐘就標(biāo)記該條數(shù)據(jù)時間戳異常預(yù)警和報(bào)表默認(rèn)剔除避免臟數(shù)據(jù)污染統(tǒng)計(jì)結(jié)果。數(shù)據(jù)質(zhì)量治理還有一層重復(fù)上報(bào)。DTU誤重連、網(wǎng)絡(luò)超時重發(fā)都會導(dǎo)致同一條數(shù)據(jù)上報(bào)兩三次。我在寫入MySQL時用了唯一約束字段組合是station_id indicator_code server_time重復(fù)數(shù)據(jù)走ON DUPLICATE KEY UPDATE直接覆蓋。這個設(shè)計(jì)簡單有效線上跑了很久都沒有重復(fù)記錄。4. Django與Flask混合開發(fā)的核心實(shí)現(xiàn)4.1 Flask端輕量接入服務(wù)的寫法Flask服務(wù)不加載Django的配置和ORM只用SQLAlchemy連接池或者甚至只用Redis操作。它負(fù)責(zé)兩件事一是接收采集協(xié)議上報(bào)的數(shù)據(jù)二是給Django內(nèi)部提供幾個簡單的健康檢查和指標(biāo)查詢延伸點(diǎn)。為什么要強(qiáng)調(diào)“不加載Django配置”因?yàn)楝F(xiàn)場會經(jīng)常改接入代碼如果Flask和Django共用一套復(fù)雜配置改一次協(xié)議分析就要重啟整個業(yè)務(wù)系統(tǒng)風(fēng)險太高。我讓Flask獨(dú)立運(yùn)行在8801端口用systemd守護(hù)和Django的8000端口完全隔離。這樣現(xiàn)場協(xié)議調(diào)整時重啟Flask在線用戶完全無感知。Flask端我順手加了一個線程池做數(shù)據(jù)清洗邏輯不重的場景直接在主線程里處理也行但一旦上報(bào)頻率高比如幾秒一條建議把MQTT回調(diào)函數(shù)的處理邏輯盡量簡化只做“解析-校驗(yàn)-入隊(duì)”三步復(fù)雜處理放到后臺線程。4.2 Django端業(yè)務(wù)模型與API設(shè)計(jì)Django這邊負(fù)責(zé)業(yè)務(wù)模型、Admin后臺和API接口。我用Django REST Framework提供面向web端的接口比如站點(diǎn)列表、實(shí)時數(shù)據(jù)、歷史曲線、預(yù)警列表、統(tǒng)計(jì)報(bào)表。JWT認(rèn)證用來保護(hù)非公開接口大屏展示類接口則單獨(dú)簽發(fā)只讀token不占用用戶登錄態(tài)。核心模型大致是這樣class Station(models.Model): code models.CharField(站點(diǎn)編碼, max_length32, uniqueTrue) name models.CharField(站點(diǎn)名稱, max_length64) longitude models.FloatField(經(jīng)度) latitude models.FloatField(緯度) section models.CharField(斷面位置, max_length128) status models.BooleanField(啟用狀態(tài), defaultTrue) class Indicator(models.Model): code models.CharField(指標(biāo)代碼, max_length16, uniqueTrue) name models.CharField(指標(biāo)名稱, max_length32) unit models.CharField(單位, max_length16) warn_min models.FloatField(預(yù)警下限, nullTrue) warn_max models.FloatField(預(yù)警上限, nullTrue) class RecordData(models.Model): station models.ForeignKey(Station, on_deletemodels.CASCADE) indicator models.ForeignKey(Indicator, on_deletemodels.CASCADE) value models.FloatField(數(shù)值) measure_time models.DateTimeField(采集時間, db_indexTrue) class Meta: unique_together (station, indicator, measure_time)這個模型設(shè)計(jì)里站點(diǎn)和指標(biāo)是外鍵關(guān)聯(lián)查詢接口用DRF的filter參數(shù)按站點(diǎn)編碼和指標(biāo)代碼過濾。實(shí)際海量查詢不要走ORM全量查我會在Service層直接寫SQL按月份動態(tài)拼表名再把結(jié)果組裝成可視化需要的JSON結(jié)構(gòu)。4.3 權(quán)限控制與后臺管理的落地Django自帶的Admin后臺非常有用管理人員可以直接維護(hù)站點(diǎn)、指標(biāo)、閾值配置。但我沒把權(quán)限開放給所有人而是分了三個角色系統(tǒng)管理員、監(jiān)測值班員、訪客。系統(tǒng)管理員可以改配置和用戶監(jiān)測值班員只能看數(shù)據(jù)和確認(rèn)預(yù)警訪客只看大屏。DRF的權(quán)限控制我用了自定義權(quán)限類思路是每個API對應(yīng)一個操作類型然后再判斷用戶角色是否有對應(yīng)權(quán)限。預(yù)警處理接口做了特殊設(shè)計(jì)值班員確認(rèn)預(yù)警后要填寫處理意見系統(tǒng)會記錄處理人和處理時間。這個功能在后來的問題追溯里幫了大忙哪條預(yù)警是誰確認(rèn)的、怎么處理的一目了然。后臺界面默認(rèn)的Django Admin樣式比較樸素我改了list_display和list_filter讓關(guān)鍵字段直接展示再配置了幾個自定義Action比如批量啟用站點(diǎn)、批量導(dǎo)出監(jiān)測數(shù)據(jù)。對內(nèi)部系統(tǒng)來說這種“夠用不花哨”的做法比硬套一個重型前端框架務(wù)實(shí)得多。5. 可視化大屏與預(yù)警聯(lián)動5.1 用ECharts畫實(shí)時曲線與站點(diǎn)地圖前端可視化我選了ECharts生態(tài)成熟、圖表類型豐富。大屏頁面主要分三塊左側(cè)放站點(diǎn)列表和關(guān)鍵指標(biāo)卡片中間是流域站點(diǎn)地圖右側(cè)是選中站點(diǎn)的實(shí)時曲線和預(yù)警滾動列表。站點(diǎn)地圖用Leaflet加OSM瓦片每個站點(diǎn)按經(jīng)緯度打點(diǎn)顏色代表當(dāng)前整體狀態(tài)綠色正常、黃色關(guān)注、紅色報(bào)警。點(diǎn)擊地圖上的站點(diǎn)標(biāo)記右側(cè)曲線圖會切換成該站點(diǎn)的數(shù)據(jù)。ECharts的實(shí)時曲線我用的是動態(tài)追加數(shù)據(jù)的方式后端WebSocket每推送一條新數(shù)據(jù)前端就把舊數(shù)據(jù)移出窗口并追加新點(diǎn)形成滾動效果。這里有個細(xì)節(jié)曲線圖的X軸不要用設(shè)備采集時間而要用服務(wù)器接收時間。因?yàn)樵O(shè)備時間可能跳變一旦跳變前端圖表會出現(xiàn)時間倒退或者空隙觀感很差。采集時間可以在Tooltip里展示不影響主趨勢的連續(xù)性。5.2 WebSocket推送與斷線重連實(shí)時性是大屏的剛需。輪詢接口也能做但5秒一次甚至3秒一次對數(shù)據(jù)庫和帶寬都不友好。我最終用WebSocket做實(shí)時推送。Flask接入服務(wù)收到新數(shù)據(jù)后把數(shù)據(jù)寫一份到Redis發(fā)布訂閱通道WebSocket服務(wù)訂閱這個通道再推送給前端。前端使用瀏覽器原生WebSocket重點(diǎn)要處理斷線重連。我用了一個簡單方案連接斷開后指數(shù)退避重連延遲從1秒開始翻倍最多30秒重連成功后立即請求一次最新數(shù)據(jù)接口把斷線期間漏掉的數(shù)據(jù)補(bǔ)回來。這樣現(xiàn)場網(wǎng)絡(luò)抖動時大屏最多缺幾十秒的數(shù)據(jù)恢復(fù)后馬上回補(bǔ)。前端訂閱邏輯大致是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服務(wù)本身是獨(dú)立進(jìn)程不掛靠在Django或者Flask進(jìn)程內(nèi)重啟WebSocket服務(wù)不會影響數(shù)據(jù)采集和業(yè)務(wù)API。部署上用Nginx做WebSocket反向代理配置Upgrade頭就行。5.3 預(yù)警規(guī)則去抖與通知預(yù)警如果做得太“靈敏”會變成狼來了。直接拿單條超閾值的數(shù)據(jù)觸發(fā)報(bào)警經(jīng)常會出現(xiàn)瞬時波動導(dǎo)致假報(bào)警。我的處理方式是同一站點(diǎn)同一指標(biāo)連續(xù)兩條數(shù)據(jù)超閾值才生成一條有效預(yù)警而且同一站點(diǎn)同一指標(biāo)在10分鐘內(nèi)重復(fù)觸發(fā)不重復(fù)告警只更新原預(yù)警的最后觸發(fā)時間。預(yù)警判斷放在APScheduler定時任務(wù)里每5分鐘跑一次掃描最近10分鐘的監(jiān)測數(shù)據(jù)用聚合SQL按站點(diǎn)和指標(biāo)分組找出超標(biāo)記錄。再和已經(jīng)存在的未處理預(yù)警做比對決定是插入新預(yù)警還是更新已有預(yù)警。這種方式比逐條判斷更高效而且杜絕了并行任務(wù)同時跑導(dǎo)致重復(fù)插入的問題。通知部分用Webhook推送到內(nèi)部IM機(jī)器人消息內(nèi)容包含站點(diǎn)名稱、指標(biāo)、當(dāng)前值、閾值、觸發(fā)時間。消息里一定要帶“確認(rèn)預(yù)警”的連接入口值班員點(diǎn)開就能跳轉(zhuǎn)到系統(tǒng)處理頁面少一步操作響應(yīng)就能快一步。6. 部署上線與線上問題排查實(shí)錄6.1 NginxGunicorn進(jìn)程守護(hù)的部署組合部署方案我用了Nginx反向代理加GunicornDjango服務(wù)跑在8000端口Flask接入服務(wù)跑在8801端口WebSocket服務(wù)獨(dú)立跑在9000端口。Nginx統(tǒng)一監(jiān)聽80/443按路徑把請求轉(zhuǎn)發(fā)到對應(yīng)服務(wù)。關(guān)鍵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數(shù)量設(shè)為CPU核心數(shù)的2倍加1超時時間設(shè)成60秒。Django接口里如果有一個慢查詢卡住不會立刻拖死其他請求。Flask接入服務(wù)我直接用了gunicorn加單worker因?yàn)樗娜蝿?wù)簡單多worker反而會導(dǎo)致MQTT回調(diào)重復(fù)處理同一個消息。進(jìn)程守護(hù)我推薦用systemd而不是簡單的nohup。每個服務(wù)寫一個unit文件設(shè)置Restartalways異常退出后自動拉起。服務(wù)日志統(tǒng)一輸出到j(luò)ournald和日志文件線上問題排查時可以秒級翻日志。6.2 性能調(diào)優(yōu)與索引優(yōu)化上線初期我遇到過查詢變慢的問題集中在監(jiān)控大屏的“最近24小時曲線”接口。排查下來發(fā)現(xiàn)是沒走索引SQL執(zhí)行了全表掃描。加了組合索引之后查詢從2秒多降到50毫秒左右這里把索引設(shè)計(jì)再強(qiáng)調(diào)一遍字段順序一定是station_id indicator_code measure_time查詢條件里有沒有站點(diǎn)、有沒有指標(biāo)、有沒有時間范圍都能命中這個索引。數(shù)據(jù)庫連接池也要調(diào)優(yōu)。Django默認(rèn)的MySQL連接池在并發(fā)上來以后容易被打滿我改成了用連接池管理并設(shè)置最大連接數(shù)60隊(duì)列大小200。同時所有批量寫入都走事務(wù)每批500條commit一次性能和穩(wěn)定性都提升不少。還有一個很容易踩的坑Django Admin列表頁如果關(guān)聯(lián)到RecordData這類大表加載時會非常慢。解決方案是不要在后臺關(guān)聯(lián)查詢監(jiān)測明細(xì)表而是單獨(dú)做一個用只讀SQL實(shí)現(xiàn)的“數(shù)據(jù)檢索”頁面按站點(diǎn)和時間范圍查詢只返回需要的數(shù)據(jù)字段。6.3 常見故障與分析速查表我把上線后遇到的典型問題整理成了一張速查表遇到類似情況可以直接對照排查?，F(xiàn)象可能原因排查方向處理辦法某個站點(diǎn)長時間無數(shù)據(jù)DTU離線、SIM卡欠費(fèi)、串口松動查看systemd日志和MQTT主題訂閱量遠(yuǎn)程重啟DTU檢查現(xiàn)場供電和串口連接數(shù)據(jù)偶發(fā)亂碼或數(shù)值突變Modbus CRC錯誤、傳感器電源不穩(wěn)看Flask日志里的原始報(bào)文增加CRC校驗(yàn)丟棄傳感器端加穩(wěn)壓電源曲線時間軸出現(xiàn)空隙設(shè)備時間跳變、網(wǎng)絡(luò)中斷對比server_time和device_time以服務(wù)器時間為準(zhǔn)斷線補(bǔ)發(fā)數(shù)據(jù)預(yù)警重復(fù)觸發(fā)去重邏輯沒生效查預(yù)警表里同一時間點(diǎn)的記錄加唯一約束和10分鐘去重窗口頁面實(shí)時數(shù)據(jù)卡住WebSocket斷連未重連看瀏覽器控制臺WebSocket狀態(tài)啟用指數(shù)退避重連并補(bǔ)拉最新數(shù)據(jù)數(shù)據(jù)庫連接數(shù)打滿連接池太小或慢查詢太多查看慢查詢?nèi)罩竞瓦B接數(shù)監(jiān)控優(yōu)化SQL索引擴(kuò)大連接池上限串口相關(guān)的故障排查看起來很低級但實(shí)際卻是最容易遇到的大問題。有一次某個站點(diǎn)上報(bào)的pH值一直在7到9之間亂跳排查了兩天最后發(fā)現(xiàn)是探頭前端被泥沙糊住了。所以線上系統(tǒng)一定要有“長時間趨勢異?！钡奶崾咀屩蛋鄦T去現(xiàn)場清洗探頭單靠程序算法是解決不了物理世界問題的。另一個值得說的經(jīng)驗(yàn)是監(jiān)控預(yù)警系統(tǒng)自身。我給Flask接入服務(wù)加了心跳接口每30秒上報(bào)一次服務(wù)狀態(tài)到Django同時內(nèi)部有一個“數(shù)據(jù)新鮮度檢查”任務(wù)如果某個站點(diǎn)超過15分鐘沒有新數(shù)據(jù)系統(tǒng)會自動生成一條設(shè)備離線告警。這個設(shè)計(jì)讓我不需要時時盯著系統(tǒng)異常發(fā)生時第一時間就能收到通知。我個人在實(shí)際操作中的體會是流域水情水質(zhì)監(jiān)測系統(tǒng)的成敗很大程度不在框架選得多先進(jìn)而在于數(shù)據(jù)鏈路是否穩(wěn)定、數(shù)據(jù)是否可信、異常是否能被發(fā)現(xiàn)。Django和Flask混搭這套組合幫我很好地平衡了接入靈活性和業(yè)務(wù)開發(fā)效率如果你也準(zhǔn)備做類似的項(xiàng)目建議先把采集鏈路和數(shù)據(jù)模型打磨扎實(shí)再考慮花哨的展示功能。最后再分享一個小技巧所有設(shè)備的通信參數(shù)和校驗(yàn)算法不要只放在代碼里一定要同步一份到項(xiàng)目文檔中現(xiàn)場人員排查問題時能省下大量溝通成本。