去年开春实验室负责人交给我一件事三间屋里二十几个工位实验台、药品柜、培养箱、冷藏柜全都需要长期盯着温度和湿度。以前依靠人手拿仪表挨个测费时间不说还经常漏记录。我最后落地的方案是给每个工位配一台带以太网口的温湿度记录仪走UDP协议把数据发到一台轻量服务器上服务器端写一个接收程序统一收包、解析、入库、告警。整套程序加起来不到两百行跑在一台退役办公小主机上稳稳定定跑了半年多。如果你也在做实验室多工位监测、机房环境监控或者仓库温湿度记录又不想上一堆重型平台这套“UDP协议 以太网记录仪 轻量服务器接收程序”的架构可以直接参考。1. 项目为什么这么设计UDP以太网方案的取舍1.1 先说说为什么排除了WiFi和TCP一开始有同事建议用WiFi温湿度传感器理由是无线路由器现成设备便宜。但真放在实验室里测了一圈就发现行不通实验台之间的金属柜体、玻璃隔断、仪器外壳都会明显衰减信号个别点位甚至放在接种间里的恒温培养箱内部WiFi信号进去只剩一格设备频繁掉线重连数据链路非常不稳定。有线以太网不存在这些问题一根网线通到底只要交换机端口正常物理链路就是稳的而且排查也直接。那为什么传输层选UDP而不是TCP很多人习惯性地认为TCP更可靠但在这个场景里UDP反而是更优解。温湿度数据包一般就十几个字节采集间隔几秒到几十秒偶发丢一包根本不影响整体曲线下一包马上就把信息补回来了。TCP有连接建立、确认重传、拥塞控制这一整套机制几十台设备同时维护长连接不仅浪费设备资源断线重连的逻辑也更复杂UDP就省事了无连接状态设备配置好目标IP和端口之后上电即发服务端只要把socket开着什么都不用管。可以这么理解TCP像挂号信UDP像明信片——挂号信流程严谨但也慢明信片轻巧偶尔丢一张无所谓反正下一张马上到。我还看重UDP的一点是扩展性一个端口可以同时接收几十上百台设备的包整个实验室网络里的任何一台授权机器只要发到一个目标端口就行服务端这边不用为每个工位单独开端口或维护连接状态多工位场景天然适配。1.2 多工位数据流的整体架构整个系统的组网其实特别简单核心就是记录仪经过网线接到实验室交换机服务器也接到同一台交换机所有设备把数据发到服务器的同一个UDP端口接收程序收到后再根据报文中携带的设备ID区分是哪个工位。具体数据流是这样的探头采集温湿度 → 记录仪本地计时、存储同时组装成UDP报文 → 交换机 → 服务器接收程序解析 → 写入SQLite数据库 → 超出阈值或设备离线时触发告警 → 可选地推送到看板或群消息。每个环节都很轻没有消息队列、没有微服务、没有复杂的容器编排整套链路里真正的核心就是那个不到两百行的接收程序。在很多项目管理经验里第一步最容易犯的错是先考虑“用什么框架”我这次反过来先定协议再定存储再反推程序。因为UDP报文小、频率低、格式固定所以Python的socket配合SQLite完全能扛住根本不需要过早引入大数据那一套。这个判断在后续几个月运行中也得到了验证。2. 硬件选型与网络环境准备2.1 选设备时我盯住的四个关键点市面上支持以太网的温湿度记录仪很多选型时我重点看四件事。第一是协议是否公开且可配置。这个最重要。有些设备只支持自己的私有平台或者必须用厂家软件才能收发数据这种直接排除。我选的设备必须支持UDP主动上报且厂商能提供完整的报文协议说明包括帧头、字段长度、字节序、校验算法。没有公开协议后续接收程序就是无米之炊。第二是传感器精度和探头形式。实验室监测一般温度精度±0.3℃、湿度精度±2%RH就够用了重点是探头要可换因为温湿度探头是消耗品时间长了会漂移能拧下来单独校准或更换能省不少成本。最好选带校准证书或支持送检的型号实验室体系审查时会用到。第三是供电方式。优先选支持PoE供电的设备一根网线同时解决数据回传和供电省去到处找插座、拉电源线的麻烦。没有PoE就选集中DC电源方案的机型尽量不要每一台配一个充电头插排很容易被保洁碰到。第四是设备ID的配置能力。多工位场景必须给每台设备分配唯一ID或IP可以通过DIP拨码、数码管菜单或上位机工具设置调试时能快速区分和变更。这一条看着简单真到现场一堆设备摆在一起时就特别重要。这里多说一句我也考虑过自己用STM32或ESP32加SHT30传感器、再用W5500模块接网线做采集器原理图都研究过成本确实低但最后放弃了。实验室项目最怕的不是硬件设计不出来而是后续校准、维修、更换时没有标准件和市场存量支持。成品记录仪坏了直接换新机器就行自己搓的板子坏了还得备料、烧固件、重新测工作量完全不在一个量级。DIY方案更适合学习或特殊定制场景常规监测我还是建议买成熟设备。2.2 交换机和网络规划建议网络这部分不用花大钱一台支持PoE供电的普通百兆交换机就够了。温湿度包每秒才几KB量级千兆纯属浪费。我这边用的是24口PoE交换机网线供电一起解决机柜里看起来很清爽。IP规划要有全局观。我给所有设备规划了固定的网段服务器和记录仪都固定在192.168.10.x服务器10.10设备从11到50一律配置静态IP不用DHCP。因为UDP是无连接的服务端根本不知道设备什么时候在线如果设备IP频繁变化排查和监控都会乱套。DHCP适合办公网不适合固定工位监测网。如果交换机支持VLAN最好再单独划一个设备VLAN把温湿度监测网络和办公室电脑网络隔离开避免广播风暴和其他人的电脑抢IP。端口方面我选的是高段位端口50000避开路由、打印机、NAS等设备的常用端口也方便防火墙规则单独放行。所有工位设备都发到同一个端口接收端通过报文里的设备ID识别工位端口配置过程非常简单。还需要注意如果后续做高可用备份可以有多个接收端同时监听同一个UDP端口设备配置成广播或组播模式时能同时给多个服务器发数据。一套采集链路喂多个消费端这是TCP很难直接做到的。2.3 接线、供电与现场踩线经验硬件安装的坑比软件多得多。我这次遇到的最大问题其实是网线质量。设备随机附带的网线特别细当时想省事直接用了结果有两个点位隔三差五丢包后来发现是水晶头簧片质量差柜门一关就接触不良。后来全部换成超五类成品线问题再没出现过。网线走线还要注意远离强电和干扰源。实验室里有离心机、大功率磁力搅拌器、低温冰箱压缩机这些设备启动时会有电磁干扰网线贴着走容易导致偶发丢包。有条件就穿金属线槽或者用屏蔽双绞线没条件至少保持30厘米以上的间距。供电方面我强烈建议每台设备的网口旁边贴一个小标签写明“设备ID、工位名称、IP地址、安装日期”。同样的信息也在交换机面板端口上贴一份。这个土办法在排查问题时极其管用不用每次抱个笔记本电脑在那查IP。另外要特别提醒记录仪本身要支持本地存储。虽然网络传输是我们这套方案的主链路但万一交换机断电、服务器重启记录仪还能把数据存在本地SD卡或Flash里事后可以补录。单靠网络实时上报就算UDP再轻量链路断了也照样收不到。3. 轻量服务器接收程序的实现3.1 接收程序到底管几件事写代码之前我先把接收程序的职责边界画清楚它只是一个“门卫”负责收件、验身、记账、叫人不做重传不做复杂队列不做登录鉴权。具体就六件事监听UDP端口持续接收报文按厂家协议解析数据提取设备ID、温度、湿度并做校验把原始设备ID映射成工位名称如“培养箱1号”“药品柜A”把解析结果写入SQLite数据库带准确时间戳检查是否越限温度上下限、湿度上下限越限就告警维护各设备最后活跃时间检测离线状态。很多人容易把这类小工具写复杂接个RabbitMQ、上Redis、搞个K8s部署其实完全没必要。这个场景的数据特征是小包、低频、单写多读。一个Python进程、一个UDP socket、一个SQLite数据库就能覆盖绝大多数实验室工位监测需求维护成本极低。3.2 先做一版最简代码跑通再说正式解析协议之前我建议你先写一个最简单的监听脚本只把收到的报文用十六进制打出来确认设备确实在发包。这个版本只要几行import socket sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind((0.0.0.0, 50000)) sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024) print(正在监听 UDP 50000 端口...) while True: data, addr sock.recvfrom(2048) print(f来自 {addr} 的报文: {data.hex()})这里有几个细节要解释。socket.SOCK_DGRAM就是UDP协议要和TCP的SOCK_STREAM区分开来。bind的IP填0.0.0.0表示监听本机所有网卡这样不管设备从哪块网卡发包都能收到别填127.0.0.1那只能收到本机自己发的测试包真实设备在网线另一端是过不来的。SO_RCVBUF我设成1MB是为了防止多个设备同时突发上报时内核缓冲区不够导致丢包这个在Windows上尤其明显默认缓冲区太小。跑起来之后如果设备配置正确你很快就能在终端里看到一串十六进制报文。这一步的意义是验证链路通没通先别急着做解析。3.3 多工位解析、入库和告警完整实现链路确认通了之后再上完整逻辑。下面这个版本是我实际在用的核心代码去掉了业务无关部分保留了完整的模式import socket import struct import sqlite3 import time import threading UDP_IP 0.0.0.0 UDP_PORT 50000 OFFLINE_SECONDS 300 DB_PATH monitor.db # 工位映射表设备ID - 名称、阈值、在线状态 devices { 1: {name: 培养箱1号, temp_min: 15.0, temp_max: 30.0, hum_min: 30.0, hum_max: 80.0, last_seen: 0, online: False}, 2: {name: 药品柜A, temp_min: 5.0, temp_max: 20.0, hum_min: 30.0, hum_max: 70.0, last_seen: 0, online: False}, # 继续按工位添加... } def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS env_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER NOT NULL, device_name TEXT NOT NULL, temperature REAL NOT NULL, humidity REAL NOT NULL, ts DATETIME DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def insert_record(dev_id, temp, humi): conn sqlite3.connect(DB_PATH) conn.execute( INSERT INTO env_data(device_id, device_name, temperature, humidity) VALUES(?,?,?,?), (dev_id, devices[dev_id][name], temp, humi) ) conn.commit() conn.close() def parse_packet(data): # 示例协议0xA5 设备ID(2字节) 温度(0.01℃,有符号) 湿度(0.01%RH) 设备状态(1字节) # 具体字段必须严格对照设备手册调整这里给出常见构型 if len(data) 8 or data[0] ! 0xA5: return None dev_id struct.unpack(H, data[1:3])[0] temp struct.unpack(h, data[3:5])[0] / 100.0 humi struct.unpack(H, data[5:7])[0] / 100.0 return dev_id, temp, humi def alert(msg): # 先输出日志后续可替换成钉钉/邮件通知 print(f[ALERT] {time.strftime(%Y-%m-%d %H:%M:%S)} {msg}) def watchdog(): while True: time.sleep(10) now time.time() for dev_id, dev in devices.items(): if now - dev[last_seen] OFFLINE_SECONDS: if dev[online]: dev[online] False alert(f设备离线: {dev[name]}设备ID {dev_id}) def main(): init_db() sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind((UDP_IP, UDP_PORT)) sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024) print(f监听 UDP {UDP_PORT} 端口工位数: {len(devices)}) threading.Thread(targetwatchdog, daemonTrue).start() while True: data, addr sock.recvfrom(2048) parsed parse_packet(data) if not parsed: print(f无法解析的报文: {data.hex()}) continue dev_id, temp, humi parsed if dev_id not in devices: print(f未知设备ID {dev_id}) continue dev devices[dev_id] dev[last_seen] time.time() if not dev[online]: dev[online] True alert(f设备恢复在线: {dev[name]}) insert_record(dev_id, temp, humi) print(f{time.strftime(%H:%M:%S)} {dev[name]}: {temp:.2f}℃ / {humi:.2f}%RH) if not (dev[temp_min] temp dev[temp_max]): alert(f{dev[name]} 温度超限: {temp:.2f}℃) if not (dev[hum_min] humi dev[hum_max]): alert(f{dev[name]} 湿度超限: {humi:.2f}%RH) if __name__ __main__: main()代码里重点看几个设计。解析协议时温度我用的是有符号短整型h因为冷链或低温冰箱会降到零下不能用无符号湿度用H。byte order用大端还是小端完全取决于设备手册这一步最容易错最好拿一个已知设备报文对照验证比如25℃应该解析成2500如果出来是640这种怪数基本就是字节序或符号位弄反了。3.4 告警、离线检测与生产补强上面代码里我埋了一个看门狗线程每10秒检查一次所有工位的最后活跃时间超过300秒没收到包就判定离线。离线检测很关键因为UDP无连接的特性决定了服务端不可能主动知道设备掉线只能靠“超时未收包”来推断。告警我这里只写了控制台打印实际使用时我建议直接接到钉钉群机器人Webhook或者邮件通知这样不用时刻盯着终端。但有一个重要坑告警要做去重和恢复。比如温度一直超限不能每收到一个包就发一次告警那样十分钟就能刷爆群消息。比较稳妥的做法是设置“连续多次越限才告警”离线告警也要在真正离线时只报一次等恢复后再报恢复。代码里我已经通过dev[online]的翻转实现了这个逻辑但越限去重需要你根据自己的阈值表再优化一下比如加一个last_alert_time。还有一个容易被忽略的点每个人的工位判定策略不同。培养箱温度波动很正常短时间超限不代表真的异常这时候可以增加滑动平均取最近五次数据的均值再判断。不要对单次采样值过于敏感。4. 调试现场模拟发包、抓包和常见坑4.1 没有设备也能先调程序模拟包与UDP调试工具我调试这套程序的时候第一批设备还没全部到货但接收程序和数据库表结构已经写完了怎么验证答案是模拟发包。只要协议手册在手就能用Python伪造一包数据import socket import struct s socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 模拟设备ID1温度25.00℃湿度50.00%RH pkt bytes([0xA5]) struct.pack(H, 1) struct.pack(h, 2500) struct.pack(H, 5000) s.sendto(pkt, (127.0.0.1, 50000))跑完这条接收程序终端里应该立刻出现“培养箱1号: 25.00℃ / 50.00%RH”的记录。用同样的方式多改几个设备ID就能把多工位识别的逻辑整个测一遍。这里先发到127.0.0.1验证本机程序逻辑没问题后再把目标换成服务器的真实IP同时把socket绑定的地址改成0.0.0.0验证跨机器链路。如果你不喜欢写代码也可以用现成的串口/网络调试工具比如“UDP网络调试助手”或“SocketTool”界面里填目标IP和端口勾选HEX发送手动输入A5 00 01 09 C4 13 88这类报文就能发。这种工具在设备厂商现场不支持改协议参数时特别有用可以直接手动发包测试服务端响应。4.2 Wireshark抓包定位问题的实用方法程序调试遇到收不到包的情况我第一个打开Wireshark把网卡选到服务器对应网口过滤器填udp.port 50000回车就能看到所有发往这个端口的报文。这一步能快速区分到底是设备没发包还是服务端没收到或者是程序解析出了问题。抓包有几个实用经验。第一关注源IP和目的IP如果源IP不是设备规划的固定地址段说明设备配的静态IP不对或者有两台设备IP冲突。第二直接看Wireshark底部的HEX区域对照协议手册逐字段人工解析一次能帮你快速发现问题。第三注意UDP长度为1473以上的报文常规以太网MTU是1500字节扣除IP头和UDP头UDP payload超过1472就会分片虽然温湿度包这么小的概率几乎为零但如果设备固件异常把日志塞进UDP包就可能看到分片分片丢失会导致整个包无法解析服务端recvfrom什么也收不到。我还习惯用iperf3做一次UDP网络打流测试判断链路本身有没有丢包。在服务器上启动iperf3 -s -p 50001在另一端跑iperf3 -u -c 服务器IP -p 50001 -b 1M -t 60如果丢包率超过0.01%就要查交换机和网线质量。这一步能提前排除很多物理层问题。4.3 常见问题速查表考虑到踩坑记录比较多这里整理成一张速查表方便你后续对照现象可能原因排查与处理终端有报文但解析失败协议字段长度/字节序与手册不一致用Wireshark对照HEX逐字节核对个别工位完全收不到数据网线松动、IP冲突、设备掉电ping设备IP查arp表检查交换机端口指示灯温度数值偶尔跳变到离谱校验不严或瞬时干扰丢弃校验失败包采用滑动平均增加突变阈值过滤运行几天后进程卡死sqlite连接未关闭、缓冲区占满确保insert_record中连接用完即close捕获异常并记录日志重启服务后内存持续上涨调试语句大量输出或线程泄漏关闭print调试线程都用daemon并控制sleep周期UDP包收不全设备发送频率过高、内核缓冲区太小调大SO_RCVBUF适当降低设备上报频率补充一个我自己的排查习惯先用ping 设备IP -t看通断再用ss -ulnp | grep 50000确认服务端确实在监听端口最后才开抓包工具。这样由链路层到传输层再到应用层一层层定位不会翻来覆去乱猜。4.4 服务端网络状态怎么看快速查看UDP监听状态Linux用ss -ulnpWindows用netstat -an | findstr 50000。如果看到那条监听的LISTENING条目说明服务端socket没问题。如果服务端启动了但端口没在监听多半是代码没跑起来或在bind阶段异常退出去看程序日志比查设备更优先。另外Windows防火墙默认会拦截外部设备的UDP入站报文程序第一次运行时通常会弹窗提醒别手滑点取消。Linux服务器如果是最小化安装iptables或firewalld也可能拦。调试时先临时放行确认正常后再细化规则。5. 让它长期稳定运行并向前延伸5.1 做成开机自启服务接收程序要长期跑肯定不能每次都手动打开终端。Linux下我用systemd做成服务配置文件放在/etc/systemd/system/tempmonitor.service里[Unit] DescriptionTemp Humidity UDP Monitor Afternetwork.target [Service] ExecStart/usr/bin/python3 /opt/tempmonitor/server.py WorkingDirectory/opt/tempmonitor Restartalways RestartSec5 Usermonitor [Install] WantedBymulti-user.target启动并设为开机自启systemctl enable --now tempmonitor.service。Restartalways能保证进程异常退出之后5秒内自动拉起这对无人值守的实验室环境特别重要。Windows机器则可以用NSSM把python.exe注册成服务nssm install TempMonitor然后在界面里填Command和Arguments同样能开机自启崩溃后自动重启。我实际部署时还加了一层watchdog不是代码里的离线检测而是系统级的每周定时检查一次服务状态如果systemd显示服务挂掉次数过多就发通知提醒我查看设备网络或供电情况。毕竟硬件环境再稳定总有些意外。5.2 数据后续怎么用对接看板与归档数据写进SQLite之后很多衍生的需求就很好做了。最简单的方案是写一个Flask小进程读同一个数据库文件在浏览器上显示最新值和历史曲线。注意SQLite支持多进程同时读但写入要尽量避免两个进程同时写同一个库所以我建议接收程序是唯一的写者其他展示程序只读。数据量大了以后要想着归档和清理。温湿度数据一天大约能产生假设20台设备、5秒一包每天约34万条记录SQLite完全能存但几百万条之后查询会变慢做趋势曲线时也要注意加索引和按时间过滤。我这里的方案是每个月归档一次关闭接收程序用SQLite的.backup命令复制出当月数据文件清空主库保留最近一个月的活跃数据。这样做既避免了数据库无限膨胀又保留了历史备查还能在实验室体系审计时快速拿出记录。如果后期要跨多个实验室、跨园区统一看板再考虑上InfluxDB这类时序数据库或者把SQLite数据通过脚本导出到统一平台。但对单实验室多工位来说SQLite足够稳妥。5.3 关于数据安全和部署边界的一点提醒最后聊几句安全边界这部分很容易被忽略。当前这套UDP服务和工位设备都在封闭的实验室局域网内正常情况下不要把服务端的UDP端口直接暴露到公网。UDP没有连接状态公网扫描器只要知道端口和报文格式可以伪造任意设备ID往你的服务器灌脏数据既污染数据库又可能触发一堆假告警。数据确实需要远程查看的话走实验室统一的安全网关做认证转发或者干脆放在内网远程时通过堡垒机登录服务器再看。反正我个人一直坚持设备网段和办公网段隔离接收服务只对可信机器开放。另外传感器本身的校准周期也要规划。电子温湿度探头漂移是常态实验室要求高的场合至少每年送检一次必要时更换探头。程序里我建议加一个“设备校准到期提醒”在设备映射表里加一个last_calibrate_date字段临期时给运维人员发提醒。这一点看起来和技术无关但在实际运行中很能体现工程完整性。最后说两个小经验。设备ID和工位位置的对应表我不仅写进程序还打印出来贴在交换机面板旁边现场排查时不用抱着笔记本电脑到处跑。另外强烈建议先把模拟包测试跑通再接真实设备一个本机模拟能在十分钟内验证整个接收、解析、入库、告警流程能省掉大量现场来回跑腿的时间。这套结构处理二十来个大大小小的工位绰绰有余未来点位翻倍也基本不用动核心逻辑改改映射表和采集周期很快就能顶上去。
UDP协议+以太网记录仪：实验室多工位温湿度监测轻量方案
去年开春实验室负责人交给我一件事三间屋里二十几个工位实验台、药品柜、培养箱、冷藏柜全都需要长期盯着温度和湿度。以前依靠人手拿仪表挨个测费时间不说还经常漏记录。我最后落地的方案是给每个工位配一台带以太网口的温湿度记录仪走UDP协议把数据发到一台轻量服务器上服务器端写一个接收程序统一收包、解析、入库、告警。整套程序加起来不到两百行跑在一台退役办公小主机上稳稳定定跑了半年多。如果你也在做实验室多工位监测、机房环境监控或者仓库温湿度记录又不想上一堆重型平台这套“UDP协议 以太网记录仪 轻量服务器接收程序”的架构可以直接参考。1. 项目为什么这么设计UDP以太网方案的取舍1.1 先说说为什么排除了WiFi和TCP一开始有同事建议用WiFi温湿度传感器理由是无线路由器现成设备便宜。但真放在实验室里测了一圈就发现行不通实验台之间的金属柜体、玻璃隔断、仪器外壳都会明显衰减信号个别点位甚至放在接种间里的恒温培养箱内部WiFi信号进去只剩一格设备频繁掉线重连数据链路非常不稳定。有线以太网不存在这些问题一根网线通到底只要交换机端口正常物理链路就是稳的而且排查也直接。那为什么传输层选UDP而不是TCP很多人习惯性地认为TCP更可靠但在这个场景里UDP反而是更优解。温湿度数据包一般就十几个字节采集间隔几秒到几十秒偶发丢一包根本不影响整体曲线下一包马上就把信息补回来了。TCP有连接建立、确认重传、拥塞控制这一整套机制几十台设备同时维护长连接不仅浪费设备资源断线重连的逻辑也更复杂UDP就省事了无连接状态设备配置好目标IP和端口之后上电即发服务端只要把socket开着什么都不用管。可以这么理解TCP像挂号信UDP像明信片——挂号信流程严谨但也慢明信片轻巧偶尔丢一张无所谓反正下一张马上到。我还看重UDP的一点是扩展性一个端口可以同时接收几十上百台设备的包整个实验室网络里的任何一台授权机器只要发到一个目标端口就行服务端这边不用为每个工位单独开端口或维护连接状态多工位场景天然适配。1.2 多工位数据流的整体架构整个系统的组网其实特别简单核心就是记录仪经过网线接到实验室交换机服务器也接到同一台交换机所有设备把数据发到服务器的同一个UDP端口接收程序收到后再根据报文中携带的设备ID区分是哪个工位。具体数据流是这样的探头采集温湿度 → 记录仪本地计时、存储同时组装成UDP报文 → 交换机 → 服务器接收程序解析 → 写入SQLite数据库 → 超出阈值或设备离线时触发告警 → 可选地推送到看板或群消息。每个环节都很轻没有消息队列、没有微服务、没有复杂的容器编排整套链路里真正的核心就是那个不到两百行的接收程序。在很多项目管理经验里第一步最容易犯的错是先考虑“用什么框架”我这次反过来先定协议再定存储再反推程序。因为UDP报文小、频率低、格式固定所以Python的socket配合SQLite完全能扛住根本不需要过早引入大数据那一套。这个判断在后续几个月运行中也得到了验证。2. 硬件选型与网络环境准备2.1 选设备时我盯住的四个关键点市面上支持以太网的温湿度记录仪很多选型时我重点看四件事。第一是协议是否公开且可配置。这个最重要。有些设备只支持自己的私有平台或者必须用厂家软件才能收发数据这种直接排除。我选的设备必须支持UDP主动上报且厂商能提供完整的报文协议说明包括帧头、字段长度、字节序、校验算法。没有公开协议后续接收程序就是无米之炊。第二是传感器精度和探头形式。实验室监测一般温度精度±0.3℃、湿度精度±2%RH就够用了重点是探头要可换因为温湿度探头是消耗品时间长了会漂移能拧下来单独校准或更换能省不少成本。最好选带校准证书或支持送检的型号实验室体系审查时会用到。第三是供电方式。优先选支持PoE供电的设备一根网线同时解决数据回传和供电省去到处找插座、拉电源线的麻烦。没有PoE就选集中DC电源方案的机型尽量不要每一台配一个充电头插排很容易被保洁碰到。第四是设备ID的配置能力。多工位场景必须给每台设备分配唯一ID或IP可以通过DIP拨码、数码管菜单或上位机工具设置调试时能快速区分和变更。这一条看着简单真到现场一堆设备摆在一起时就特别重要。这里多说一句我也考虑过自己用STM32或ESP32加SHT30传感器、再用W5500模块接网线做采集器原理图都研究过成本确实低但最后放弃了。实验室项目最怕的不是硬件设计不出来而是后续校准、维修、更换时没有标准件和市场存量支持。成品记录仪坏了直接换新机器就行自己搓的板子坏了还得备料、烧固件、重新测工作量完全不在一个量级。DIY方案更适合学习或特殊定制场景常规监测我还是建议买成熟设备。2.2 交换机和网络规划建议网络这部分不用花大钱一台支持PoE供电的普通百兆交换机就够了。温湿度包每秒才几KB量级千兆纯属浪费。我这边用的是24口PoE交换机网线供电一起解决机柜里看起来很清爽。IP规划要有全局观。我给所有设备规划了固定的网段服务器和记录仪都固定在192.168.10.x服务器10.10设备从11到50一律配置静态IP不用DHCP。因为UDP是无连接的服务端根本不知道设备什么时候在线如果设备IP频繁变化排查和监控都会乱套。DHCP适合办公网不适合固定工位监测网。如果交换机支持VLAN最好再单独划一个设备VLAN把温湿度监测网络和办公室电脑网络隔离开避免广播风暴和其他人的电脑抢IP。端口方面我选的是高段位端口50000避开路由、打印机、NAS等设备的常用端口也方便防火墙规则单独放行。所有工位设备都发到同一个端口接收端通过报文里的设备ID识别工位端口配置过程非常简单。还需要注意如果后续做高可用备份可以有多个接收端同时监听同一个UDP端口设备配置成广播或组播模式时能同时给多个服务器发数据。一套采集链路喂多个消费端这是TCP很难直接做到的。2.3 接线、供电与现场踩线经验硬件安装的坑比软件多得多。我这次遇到的最大问题其实是网线质量。设备随机附带的网线特别细当时想省事直接用了结果有两个点位隔三差五丢包后来发现是水晶头簧片质量差柜门一关就接触不良。后来全部换成超五类成品线问题再没出现过。网线走线还要注意远离强电和干扰源。实验室里有离心机、大功率磁力搅拌器、低温冰箱压缩机这些设备启动时会有电磁干扰网线贴着走容易导致偶发丢包。有条件就穿金属线槽或者用屏蔽双绞线没条件至少保持30厘米以上的间距。供电方面我强烈建议每台设备的网口旁边贴一个小标签写明“设备ID、工位名称、IP地址、安装日期”。同样的信息也在交换机面板端口上贴一份。这个土办法在排查问题时极其管用不用每次抱个笔记本电脑在那查IP。另外要特别提醒记录仪本身要支持本地存储。虽然网络传输是我们这套方案的主链路但万一交换机断电、服务器重启记录仪还能把数据存在本地SD卡或Flash里事后可以补录。单靠网络实时上报就算UDP再轻量链路断了也照样收不到。3. 轻量服务器接收程序的实现3.1 接收程序到底管几件事写代码之前我先把接收程序的职责边界画清楚它只是一个“门卫”负责收件、验身、记账、叫人不做重传不做复杂队列不做登录鉴权。具体就六件事监听UDP端口持续接收报文按厂家协议解析数据提取设备ID、温度、湿度并做校验把原始设备ID映射成工位名称如“培养箱1号”“药品柜A”把解析结果写入SQLite数据库带准确时间戳检查是否越限温度上下限、湿度上下限越限就告警维护各设备最后活跃时间检测离线状态。很多人容易把这类小工具写复杂接个RabbitMQ、上Redis、搞个K8s部署其实完全没必要。这个场景的数据特征是小包、低频、单写多读。一个Python进程、一个UDP socket、一个SQLite数据库就能覆盖绝大多数实验室工位监测需求维护成本极低。3.2 先做一版最简代码跑通再说正式解析协议之前我建议你先写一个最简单的监听脚本只把收到的报文用十六进制打出来确认设备确实在发包。这个版本只要几行import socket sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind((0.0.0.0, 50000)) sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024) print(正在监听 UDP 50000 端口...) while True: data, addr sock.recvfrom(2048) print(f来自 {addr} 的报文: {data.hex()})这里有几个细节要解释。socket.SOCK_DGRAM就是UDP协议要和TCP的SOCK_STREAM区分开来。bind的IP填0.0.0.0表示监听本机所有网卡这样不管设备从哪块网卡发包都能收到别填127.0.0.1那只能收到本机自己发的测试包真实设备在网线另一端是过不来的。SO_RCVBUF我设成1MB是为了防止多个设备同时突发上报时内核缓冲区不够导致丢包这个在Windows上尤其明显默认缓冲区太小。跑起来之后如果设备配置正确你很快就能在终端里看到一串十六进制报文。这一步的意义是验证链路通没通先别急着做解析。3.3 多工位解析、入库和告警完整实现链路确认通了之后再上完整逻辑。下面这个版本是我实际在用的核心代码去掉了业务无关部分保留了完整的模式import socket import struct import sqlite3 import time import threading UDP_IP 0.0.0.0 UDP_PORT 50000 OFFLINE_SECONDS 300 DB_PATH monitor.db # 工位映射表设备ID - 名称、阈值、在线状态 devices { 1: {name: 培养箱1号, temp_min: 15.0, temp_max: 30.0, hum_min: 30.0, hum_max: 80.0, last_seen: 0, online: False}, 2: {name: 药品柜A, temp_min: 5.0, temp_max: 20.0, hum_min: 30.0, hum_max: 70.0, last_seen: 0, online: False}, # 继续按工位添加... } def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS env_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER NOT NULL, device_name TEXT NOT NULL, temperature REAL NOT NULL, humidity REAL NOT NULL, ts DATETIME DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def insert_record(dev_id, temp, humi): conn sqlite3.connect(DB_PATH) conn.execute( INSERT INTO env_data(device_id, device_name, temperature, humidity) VALUES(?,?,?,?), (dev_id, devices[dev_id][name], temp, humi) ) conn.commit() conn.close() def parse_packet(data): # 示例协议0xA5 设备ID(2字节) 温度(0.01℃,有符号) 湿度(0.01%RH) 设备状态(1字节) # 具体字段必须严格对照设备手册调整这里给出常见构型 if len(data) 8 or data[0] ! 0xA5: return None dev_id struct.unpack(H, data[1:3])[0] temp struct.unpack(h, data[3:5])[0] / 100.0 humi struct.unpack(H, data[5:7])[0] / 100.0 return dev_id, temp, humi def alert(msg): # 先输出日志后续可替换成钉钉/邮件通知 print(f[ALERT] {time.strftime(%Y-%m-%d %H:%M:%S)} {msg}) def watchdog(): while True: time.sleep(10) now time.time() for dev_id, dev in devices.items(): if now - dev[last_seen] OFFLINE_SECONDS: if dev[online]: dev[online] False alert(f设备离线: {dev[name]}设备ID {dev_id}) def main(): init_db() sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind((UDP_IP, UDP_PORT)) sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024) print(f监听 UDP {UDP_PORT} 端口工位数: {len(devices)}) threading.Thread(targetwatchdog, daemonTrue).start() while True: data, addr sock.recvfrom(2048) parsed parse_packet(data) if not parsed: print(f无法解析的报文: {data.hex()}) continue dev_id, temp, humi parsed if dev_id not in devices: print(f未知设备ID {dev_id}) continue dev devices[dev_id] dev[last_seen] time.time() if not dev[online]: dev[online] True alert(f设备恢复在线: {dev[name]}) insert_record(dev_id, temp, humi) print(f{time.strftime(%H:%M:%S)} {dev[name]}: {temp:.2f}℃ / {humi:.2f}%RH) if not (dev[temp_min] temp dev[temp_max]): alert(f{dev[name]} 温度超限: {temp:.2f}℃) if not (dev[hum_min] humi dev[hum_max]): alert(f{dev[name]} 湿度超限: {humi:.2f}%RH) if __name__ __main__: main()代码里重点看几个设计。解析协议时温度我用的是有符号短整型h因为冷链或低温冰箱会降到零下不能用无符号湿度用H。byte order用大端还是小端完全取决于设备手册这一步最容易错最好拿一个已知设备报文对照验证比如25℃应该解析成2500如果出来是640这种怪数基本就是字节序或符号位弄反了。3.4 告警、离线检测与生产补强上面代码里我埋了一个看门狗线程每10秒检查一次所有工位的最后活跃时间超过300秒没收到包就判定离线。离线检测很关键因为UDP无连接的特性决定了服务端不可能主动知道设备掉线只能靠“超时未收包”来推断。告警我这里只写了控制台打印实际使用时我建议直接接到钉钉群机器人Webhook或者邮件通知这样不用时刻盯着终端。但有一个重要坑告警要做去重和恢复。比如温度一直超限不能每收到一个包就发一次告警那样十分钟就能刷爆群消息。比较稳妥的做法是设置“连续多次越限才告警”离线告警也要在真正离线时只报一次等恢复后再报恢复。代码里我已经通过dev[online]的翻转实现了这个逻辑但越限去重需要你根据自己的阈值表再优化一下比如加一个last_alert_time。还有一个容易被忽略的点每个人的工位判定策略不同。培养箱温度波动很正常短时间超限不代表真的异常这时候可以增加滑动平均取最近五次数据的均值再判断。不要对单次采样值过于敏感。4. 调试现场模拟发包、抓包和常见坑4.1 没有设备也能先调程序模拟包与UDP调试工具我调试这套程序的时候第一批设备还没全部到货但接收程序和数据库表结构已经写完了怎么验证答案是模拟发包。只要协议手册在手就能用Python伪造一包数据import socket import struct s socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 模拟设备ID1温度25.00℃湿度50.00%RH pkt bytes([0xA5]) struct.pack(H, 1) struct.pack(h, 2500) struct.pack(H, 5000) s.sendto(pkt, (127.0.0.1, 50000))跑完这条接收程序终端里应该立刻出现“培养箱1号: 25.00℃ / 50.00%RH”的记录。用同样的方式多改几个设备ID就能把多工位识别的逻辑整个测一遍。这里先发到127.0.0.1验证本机程序逻辑没问题后再把目标换成服务器的真实IP同时把socket绑定的地址改成0.0.0.0验证跨机器链路。如果你不喜欢写代码也可以用现成的串口/网络调试工具比如“UDP网络调试助手”或“SocketTool”界面里填目标IP和端口勾选HEX发送手动输入A5 00 01 09 C4 13 88这类报文就能发。这种工具在设备厂商现场不支持改协议参数时特别有用可以直接手动发包测试服务端响应。4.2 Wireshark抓包定位问题的实用方法程序调试遇到收不到包的情况我第一个打开Wireshark把网卡选到服务器对应网口过滤器填udp.port 50000回车就能看到所有发往这个端口的报文。这一步能快速区分到底是设备没发包还是服务端没收到或者是程序解析出了问题。抓包有几个实用经验。第一关注源IP和目的IP如果源IP不是设备规划的固定地址段说明设备配的静态IP不对或者有两台设备IP冲突。第二直接看Wireshark底部的HEX区域对照协议手册逐字段人工解析一次能帮你快速发现问题。第三注意UDP长度为1473以上的报文常规以太网MTU是1500字节扣除IP头和UDP头UDP payload超过1472就会分片虽然温湿度包这么小的概率几乎为零但如果设备固件异常把日志塞进UDP包就可能看到分片分片丢失会导致整个包无法解析服务端recvfrom什么也收不到。我还习惯用iperf3做一次UDP网络打流测试判断链路本身有没有丢包。在服务器上启动iperf3 -s -p 50001在另一端跑iperf3 -u -c 服务器IP -p 50001 -b 1M -t 60如果丢包率超过0.01%就要查交换机和网线质量。这一步能提前排除很多物理层问题。4.3 常见问题速查表考虑到踩坑记录比较多这里整理成一张速查表方便你后续对照现象可能原因排查与处理终端有报文但解析失败协议字段长度/字节序与手册不一致用Wireshark对照HEX逐字节核对个别工位完全收不到数据网线松动、IP冲突、设备掉电ping设备IP查arp表检查交换机端口指示灯温度数值偶尔跳变到离谱校验不严或瞬时干扰丢弃校验失败包采用滑动平均增加突变阈值过滤运行几天后进程卡死sqlite连接未关闭、缓冲区占满确保insert_record中连接用完即close捕获异常并记录日志重启服务后内存持续上涨调试语句大量输出或线程泄漏关闭print调试线程都用daemon并控制sleep周期UDP包收不全设备发送频率过高、内核缓冲区太小调大SO_RCVBUF适当降低设备上报频率补充一个我自己的排查习惯先用ping 设备IP -t看通断再用ss -ulnp | grep 50000确认服务端确实在监听端口最后才开抓包工具。这样由链路层到传输层再到应用层一层层定位不会翻来覆去乱猜。4.4 服务端网络状态怎么看快速查看UDP监听状态Linux用ss -ulnpWindows用netstat -an | findstr 50000。如果看到那条监听的LISTENING条目说明服务端socket没问题。如果服务端启动了但端口没在监听多半是代码没跑起来或在bind阶段异常退出去看程序日志比查设备更优先。另外Windows防火墙默认会拦截外部设备的UDP入站报文程序第一次运行时通常会弹窗提醒别手滑点取消。Linux服务器如果是最小化安装iptables或firewalld也可能拦。调试时先临时放行确认正常后再细化规则。5. 让它长期稳定运行并向前延伸5.1 做成开机自启服务接收程序要长期跑肯定不能每次都手动打开终端。Linux下我用systemd做成服务配置文件放在/etc/systemd/system/tempmonitor.service里[Unit] DescriptionTemp Humidity UDP Monitor Afternetwork.target [Service] ExecStart/usr/bin/python3 /opt/tempmonitor/server.py WorkingDirectory/opt/tempmonitor Restartalways RestartSec5 Usermonitor [Install] WantedBymulti-user.target启动并设为开机自启systemctl enable --now tempmonitor.service。Restartalways能保证进程异常退出之后5秒内自动拉起这对无人值守的实验室环境特别重要。Windows机器则可以用NSSM把python.exe注册成服务nssm install TempMonitor然后在界面里填Command和Arguments同样能开机自启崩溃后自动重启。我实际部署时还加了一层watchdog不是代码里的离线检测而是系统级的每周定时检查一次服务状态如果systemd显示服务挂掉次数过多就发通知提醒我查看设备网络或供电情况。毕竟硬件环境再稳定总有些意外。5.2 数据后续怎么用对接看板与归档数据写进SQLite之后很多衍生的需求就很好做了。最简单的方案是写一个Flask小进程读同一个数据库文件在浏览器上显示最新值和历史曲线。注意SQLite支持多进程同时读但写入要尽量避免两个进程同时写同一个库所以我建议接收程序是唯一的写者其他展示程序只读。数据量大了以后要想着归档和清理。温湿度数据一天大约能产生假设20台设备、5秒一包每天约34万条记录SQLite完全能存但几百万条之后查询会变慢做趋势曲线时也要注意加索引和按时间过滤。我这里的方案是每个月归档一次关闭接收程序用SQLite的.backup命令复制出当月数据文件清空主库保留最近一个月的活跃数据。这样做既避免了数据库无限膨胀又保留了历史备查还能在实验室体系审计时快速拿出记录。如果后期要跨多个实验室、跨园区统一看板再考虑上InfluxDB这类时序数据库或者把SQLite数据通过脚本导出到统一平台。但对单实验室多工位来说SQLite足够稳妥。5.3 关于数据安全和部署边界的一点提醒最后聊几句安全边界这部分很容易被忽略。当前这套UDP服务和工位设备都在封闭的实验室局域网内正常情况下不要把服务端的UDP端口直接暴露到公网。UDP没有连接状态公网扫描器只要知道端口和报文格式可以伪造任意设备ID往你的服务器灌脏数据既污染数据库又可能触发一堆假告警。数据确实需要远程查看的话走实验室统一的安全网关做认证转发或者干脆放在内网远程时通过堡垒机登录服务器再看。反正我个人一直坚持设备网段和办公网段隔离接收服务只对可信机器开放。另外传感器本身的校准周期也要规划。电子温湿度探头漂移是常态实验室要求高的场合至少每年送检一次必要时更换探头。程序里我建议加一个“设备校准到期提醒”在设备映射表里加一个last_calibrate_date字段临期时给运维人员发提醒。这一点看起来和技术无关但在实际运行中很能体现工程完整性。最后说两个小经验。设备ID和工位位置的对应表我不仅写进程序还打印出来贴在交换机面板旁边现场排查时不用抱着笔记本电脑到处跑。另外强烈建议先把模拟包测试跑通再接真实设备一个本机模拟能在十分钟内验证整个接收、解析、入库、告警流程能省掉大量现场来回跑腿的时间。这套结构处理二十来个大大小小的工位绰绰有余未来点位翻倍也基本不用动核心逻辑改改映射表和采集周期很快就能顶上去。