接手了一个机房改造的小项目过程中最有意思的部分是把一台 UPS 的数据同时灌给两套完全独立的监控平台。一套是机房一直在用的动环监控系统走的是工业领域常见的 SNMP 轮询协议格式、OID 全是现成的另一套是我们自己搭的 Web 可视化看板要求用 WebSocket 实时推送页面刷新延迟不能超过两秒。两台系统各有各的使用场景谁也没法迁就谁最终我做了个双协议并行采集方案同一个数据源一条链路按老平台的协议格式推送另一条链路走 WebSocket 推给新看板两条通道互不干扰。这篇就把设计思路、核心代码、关键参数和踩过的坑全部整理出来给同样被多平台数据对接折磨的朋友们一个可复用的参考。1. 项目背景与核心需求拆解1.1 两套平台为什么都不能被替换先说说这个项目的实际场景。机房里有台山特的 UPS专门给核心网络设备和服务器供电。原有动环监控系统已经稳定跑了好几年值班人员平时看的是这套告警短信、电话联动也都接在这套上它读 UPS 数据的方式是标准的 SNMP 轮询间隔大概 30 秒。这套系统不能说换就换等保材料、运维流程、人员习惯全都绑定在上面。但同时管理层又提出要看实时状态大屏要求数据刷新速度快、界面好看、能叠加趋势曲线。传统动环系统的大屏能力实在有限我们自己用 Django 开发了一套 Web 监控看板前端的实时数据更新准备用 WebSocket 来做。这里就出现了一个需求差的矛盾老系统要继续用 SNMP 轮询新系统要走 WebSocket 主动推送。数据源头只有一台 UPS但两个消费端要的是两种完全不同的投递方式。这就是整个方案要解决的原始出发点。1.2 真正要解决的四个硬需求这个项目的需求看起来简单就是“把 UPS 数据推送两份”但真正深入拆解之后核心的硬指标其实有四条缺一条后面都会出事故。第一条是数据同源性。两套监控平台必须看到同一台 UPS 在同一时刻的数据不能一个显示 220.5V另一个显示 219.8V否则运维人员会认为是设备出了问题互相扯皮。解决方法是内部维护一个统一的采集快照所有下游通道都从同一个快照里取数。第二条是故障隔离性。两条推送链路要独立运行通道 A 因为网络原因挂掉了通道 B 不能跟着一起挂更不能因为一个通道的异常导致采集主进程崩溃。这一点在代码结构上要强制拆分每条通道用独立的异常处理。第三条是延迟要求。老平台的 SNMP 轮询周期可以 30 秒一次没关系但 Web 看板要求的是准实时我给自己定的目标是推送延迟不超过 2 秒。现在普遍的做法是 WebSocket 长连接服务端一旦采集到新数据就立刻往客户端推。第四条是可扩展性。虽然这次只有两个平台但保不齐后续还要加第三套比如把数据同步给上级管理平台。所以方案设计上必须留有扩展接口新增一个平台等于新增一个适配器而不是推翻重来。2. 总体设计数据模型先行双通道分流2.1 为什么不能做“串行转发”的链路结构动手之前我先把两种方案放在一起比较过。第一种是串行结构UPS 的采集进程先把数据读进来存到数据库然后由一个统一的代理服务分别往两个平台转发。这个方案代码少、逻辑直观但致命问题在于存在明显的单点瓶颈。采集进程、数据库、代理服务任何一环出问题两个平台的监控数据都会中断。而且老平台的 SNMP 轮询机制和新平台的 WebSocket 推送机制对时效性的要求不同串行结构会把二者绑死整体响应速度会被最慢的一环拖累。第二种就是最终选用的并行分流结构。采集层只负责一件事从 UPS 把原始数据读出来解析成标准模型生成一个带有时间戳的数据快照。之后这个快照同时进入两个独立的推送通道每个通道有自己独立的连接池、发送队列和异常处理逻辑。这就好比水管里要先装一个三通接头而不是把两个水桶串在同一根管子上。三通的好处很明显一个出口堵住了另外一个出口照样流水。2.2 统一内部数据模型是并行方案的地基多平台对接最容易犯的错误是“一个平台写一套解析代码”两个平台就写两遍逻辑重复、字段不一致、改起来想死。这次我反过来先定义一个内部统一的数据模型所有从 UPS 采集到的原始数据都归一化到这个模型里再在这个模型基础上去适配不同的协议。内部模型长这样字段不多但已经是实际运行验证过的device_idUPS 设备唯一标识比如“UPS-ROOM-01”collect_time采集时间戳统一用设备本地时间精确到秒input_voltage输入电压单位伏特output_voltage输出电压单位伏特output_frequency输出频率单位赫兹battery_capacity电池剩余容量百分比battery_voltage电池组电压单位伏特estimated_minutes电池后备剩余时间单位分钟status_code设备运行状态码1 代表正常0 代表旁路-1 代表通信中断snapshot_id快照序号每次采集递增为什么要用统一模型因为两个平台对数据的表示方式各不相同。老平台可能喜欢直接用 MIB 里的 OID 值新平台需要的是 JSON 字段。如果我在采集层就直接往两种协议上靠那采集层的逻辑会被协议细节污染后面加协议的时候头大。所以先出模型、再做映射这是整个双协议并行方案的地基。2.3 技术栈选型与各层分工这个项目我采用的是 Python 技术栈。采集层使用pysnmp和pymodbus库分别对接 UPS 的 SNMP 和 Modbus 接口后端服务使用 Django实时推送用 Channels 的 WebSocket 能力Redis 作为 channel layer 的消息中间件采集任务跑在一个独立进程中用 Django management command 方式启动由 supervisor 守护。采集运行频率我设置为每 2 秒执行一次保证 Web 看板的实时性同时又不会对 UPS 自身的网络接口造成太大负担。数据快照生成后分别交给通道 A 和通道 B 的发送器。每个发送器内部有独立的重试机制和队列缓存这是故障隔离的关键。3. 先把 UPS 数据搞到手采集层的实现3.1 用 SNMP 读取 UPS 的标准 MIB 数据绝大多数 UPS 都支持 SNMP 协议只要机器带网络管理卡就一定能通过 OID 读到数据。这里有个背景知识UPS 的 SNMP OID 遵循 RFC 1628 定义的 UPS MIB常见的基础信息 OID 是固定的包括设备型号、输入输出电压频率、电池状态这些。而不同厂商的 UPS 还会在私有节点下挂一些附加的扩展数据一般从1.3.6.1.4.1开头的企业私有分支下找。实际采集时我用的是pysnmp库。这个库用起来稍微有点啰嗦但胜在稳定生产环境跑了很久没出过兼容性问题。下面是一段标准的 SNMP GET 请求示例from pysnmp.hlapi import * def snmp_get(ip, community, oid): iterator getCmd( SnmpEngine(), CommunityData(community, mpModel0), UdpTransportTarget((ip, 161), timeout2, retries1), ContextData(), ObjectType(ObjectIdentity(oid)), ) error_indication, error_status, error_index, var_binds next(iterator) if error_indication: raise RuntimeError(fSNMP 通信失败: {error_indication}) return var_binds[0][1].prettyPrint()关键 OID 我列一个表格都是 UPS MIB 里的标准节点前面都带1.3.6.1.2.1.33前缀这是 UPS MIB 的根节点描述OID厂商名称1.3.6.1.2.1.33.1.1.1.0设备型号1.3.6.1.2.1.33.1.1.2.0输入电压1.4.1.0代表单相输入多相要遍历子节点输出电压1.3.6.1.2.1.33.1.4.2.0电池容量百分比1.3.6.1.2.1.33.1.2.3.0后备剩余时间分钟1.3.6.1.2.1.33.1.2.4.0电池组电压1.3.6.1.2.1.33.1.2.5.0UPS 运行状态1.3.6.1.2.1.33.1.1.3.0如果 UPS 有扩展卡或者品牌私有 MIB还可能需要去翻厂家的 MIB 文件表格里的 OID 是最通用的覆盖市面上大多数型号。实测下来要注意部分老设备的 SNMP Agent 只支持SNMPv2c所以我在CommunityData里特意指定了mpModel0避免版本不兼容导致的通信失败。3.2 备选方案Modbus 寄存器采集有些 UPS 管理卡没有开启 SNMP但提供 Modbus 协议接口。这个在工业现场也不少尤其是那些通过串口或者 TCP 方式接入现场控制柜的 UPS。读取 Modbus 数据我用的是pymodbus库走 Modbus TCP 协议默认端口是 502。Modbus 方式的麻烦在于寄存器地址完全由厂商自定义不像 SNMP 那样有相对统一的标准。厂家手册里通常会给出一个寄存器映射表比如输入电压存在 0 号寄存器输出电压存在 2 号寄存器等等。我这边碰到过一个大牌子和小牌子 UPS 的寄存器映射完全不一样所以代码里做成了配置驱动的方式每台 UPS 对应的寄存器地址全部写到一个配置字典里换设备只需要改配置不用改代码。示例代码如下from pymodbus.client import ModbusTcpClient def read_ups_modbus(ip, port502, unit_id1, register_map: dict): client ModbusTcpClient(ip, portport, timeout3) if not client.connect(): raise RuntimeError(Modbus TCP 连接失败) result {} for key, addr in register_map.items(): rr client.read_input_registers(addressaddr, count1, unitunit_id) if rr.isError(): result[key] None else: raw rr.registers[0] result[key] raw client.close() return result这里有几个注意事项。第一寄存器数据有可能是无符号整数也有可能是带符号整数具体要看厂家手册第二有些设备一个物理量占两个寄存器高位和低位要自己做拼装第三读寄存器频率不能太高不然有可能导致 UPS 通信卡死机实测下来 2 秒一次的轮询是这个设备能接受的极限了。3.3 数据归一化与采集节奏控制不管是 SNMP 还是 Modbus 拿回来的数据第一步都是归一化处理。电压、电流、频率这种模拟量厂家输出的原始值是整数往往需要除以一个缩放系数才是真实值比如 SNMP 返回 2205实际上代表 220.5V。归一化的时候就统一做换算最终存进内部模型。采集节奏方面我踩过一个坑UPS 电压的瞬时值会小幅波动直接推送给前端会导致曲线看起来像锯齿。所以我加了平滑逻辑连续取 3 次采集值做滑动平均平均值进入数据快照。这样既不会过度失真又能让曲线变得顺滑。还有一个必须处理的情况是采集失败。UPS 通信卡偶尔会有瞬间无响应网络继电器跳一下也会导致超时。我的处理方式是采集失败时不清空上一次的正常数据将状态标记为“通信中断”数据快照继续带着上一次的正常值生成但status_code会被标记成-1。这样两个平台都能及时感知到通信异常不会眼睁睁看着监控数据变成空值或者直接报错。4. 通道 A按老平台的协议格式稳定推送4.1 先搞清楚老平台到底需要哪种协议老动环平台的对接方式直接决定了通道 A 的技术路线。最常见的是三种情况平台主动来轮询、平台接收 SNMP Trap 上报、平台订阅 MQTT。这个项目里的老平台是比较传统的那一类它本身会按照配置的 SNMP 参数周期性地来采集 UPS 数据但它要求设备端有一个稳定的 SNMP Agent 一直开着。换句话说它不接收新系统主动推送的数据它要自己去“问”设备。这就带来一个有意思的问题UPS 只有一个内置的 SNMP Agent老平台已经在轮询它了如果我这边再用自己的采集进程去读同一台 UPS两台设备同时访问虽然一般不会冲突但多多少少会增加负担。更稳妥的做法是把老平台的轮询目标指向我新搭建的一台数据转换服务由这台服务“伪装”成 UPS后端数据源则是我采集到的统一模型。老平台完全感知不到后端的变化它只知道自己还能正常读到 UPS 的 OID。4.2 用自定义 SNMP Agent 模拟 UPS 设备这个模块是通道 A 的核心。我用 Python 的pysnmp库写了一个轻量的 Agent 服务监听 161 端口维护一张 OID 到数据快照字段的映射表。老平台来查询时Agent 就返回当前快照里对应的值。映射逻辑大致是这样OID_MAP { 1.3.6.1.2.1.33.1.1.2.0: lambda d: d[device_model], 1.3.6.1.2.1.33.1.4.1.0: lambda d: d[input_voltage], 1.3.6.1.2.1.33.1.4.2.0: lambda d: d[output_voltage], 1.3.6.1.2.1.33.1.2.3.0: lambda d: d[battery_capacity], 1.3.6.1.2.1.33.1.2.4.0: lambda d: d[estimated_minutes], 1.3.6.1.2.1.33.1.2.5.0: lambda d: d[battery_voltage], }这里有个经验数据快照更新之后Agent 内部的 OID 响应值必须立即同步替换。所以采集进程和 Agent 之间要共享一份内存中的快照对象我用一个带锁的全局字典来维护。老平台每次来查询Agent 直接从字典中取最新值返回不涉及数据库读操作响应速度很快。还有一个细节容易被忽略就是老平台的轮询间隔如果很短比如每 30 秒查一次那 Agent 的日志里会持续刷大量的 GET 请求。建议在 Agent 层做访问日志采样只需记录异常响应和连接错误避免日志文件爆炸式增长。4.3 通道 A 的稳定性保障细节老平台对接最怕的就是“协议没问题但时不时断一下”。这个问题的根源往往是 Agent 服务进程不稳定或者监听端口被系统防火墙拦截。我总结了三个保障要点。第一Agent 必须由进程守护工具托管比如 supervisor。这个守护进程负责拉起、重启、探活哪怕 Agent 崩溃了半分钟内也能自动恢复。第二端口和防火墙要提前确认。SNMP 用的是 UDP 161 端口很多服务器默认防火墙只放行 TCPUDP 161 没放行老平台自然访问不到。第三Agent 的返回值不能出现空值。宁可返回上一次的正常值也不能返回 null否则老平台可能会判定协议异常触发告警。5. 通道 BDjango WebSocket 实时推送到前端看板5.1 为什么实时推送一定要用 WebSocket新平台的技术选型其实在项目初期有过争议。有人提议用 HTTP 轮询前端每两秒 GET 一次接口拿最新数据代码简单Django 原生就能搞定。但我最终坚持选了 WebSocket核心原因是轮询机制有三个难以忍受的问题。首先是无效请求太多。前端每两秒请求一次哪怕 UPS 数据毫无变化服务器也要重复处理数十个请求对一个小型监控系统来说是巨大的带宽浪费。其次是延迟不可控。轮询间隔是两秒数据从采集到显示平均要经过一秒钟的等待加上网络传输和前端渲染感知延迟比 WebSocket 明显大。第三是连接开销大。大量短连接频繁建立和销毁对服务端的 TCP 资源不友好。WebSocket 的模型是长连接加服务端主动推送。前端页面只需和 Django 之间建立一条 WebSocket 连接服务端一旦从采集进程拿到新的数据快照就立即把数据包推给所有在线的客户端不需要客户端反复来问延迟可控在 1 秒以内。这也是“后台有数据前端推送”这个需求最顺手的实现路径。5.2 Django Channels 的接入与消息路由配置Django 框架原生走的是 HTTP 同步请求处理模型要支持 WebSocket就必须引入 Channels 插件。Channels 把 Django 扩展为 ASGI 应用让 Django 能够处理异步协议。配置上分三步安装依赖、设置 ASGI 路由、编写消费者。第一步安装依赖pip install channels channels-redis daphne第二步在settings.py中完成基础配置。这里我要强调 channel layer 的配置它决定了 WebSocket 消息如何在多个 worker 之间共享和路由。生产环境我使用 Redis 作为 channel layer 的存储后端ASGI_APPLICATION ups_monitor.asgi.application CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: {hosts: [(127.0.0.1, 6379)]}, } }asgi.py文件里注册 WebSocket 路由客户端的订阅路径统一放在/ws/ups/下from django.urls import re_path from channels.routing import ProtocolTypeRouter, URLRouter from ups_monitor.consumers import UPSDataConsumer websocket_urlpatterns [ re_path(rws/ups/$, UPSDataConsumer.as_asgi()), ] application ProtocolTypeRouter({ websocket: URLRouter(websocket_urlpatterns), })5.3 消费者与“有数据就推”的核心逻辑消费者类是整个通道 B 的枢纽。每个打开监控页面的浏览器都会建立一个 WebSocket 连接对应一个消费者实例。消费者在connect阶段加入一个名为ups_realtime的频道组这样服务端广播一条消息所有在线客户端都能同时收到。具体实现如下import json from channels.generic.websocket import AsyncWebsocketConsumer class UPSDataConsumer(AsyncWebsocketConsumer): async def connect(self): await self.channel_layer.group_add(ups_realtime, self.channel_name) await self.accept() await self.send(text_datajson.dumps( {type: connected, message: UPS 实时数据通道已建立} )) async def disconnect(self, code): await self.channel_layer.group_discard(ups_realtime, self.channel_name) async def ups_data_push(self, event): payload event[payload] await self.send(text_datajson.dumps(payload))真正实现“后台有数据就推”的是 group 广播的触发方。采集进程拿到数据快照之后通过 channel layer 的group_send方法把数据包发送到整个ups_realtime组from channels.layers import get_channel_layer import asyncio async def push_to_websocket(channel_layer, snapshot): await channel_layer.group_send( ups_realtime, { type: ups_data_push, payload: snapshot, } )这一段是整个方案通道 B 的核心。再用一个 while 循环把所有逻辑串起来采集生成快照推给 WebSocket 组休眠继续下一轮。循环跑在 Django management command 里和通道 A 完全分离。启动命令我写在base_command里方便用 supervisor 托管from django.core.management.base import BaseCommand from channels.layers import get_channel_layer import asyncio class Command(BaseCommand): def handle(self, *args, **options): loop asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(self.push_loop()) async def push_loop(self): channel_layer get_channel_layer() while True: snapshot await async_collect_ups() await push_to_websocket(channel_layer, snapshot) await asyncio.sleep(2)前端 JavaScript 侧连接 WebSocket 的代码也不复杂。收到消息后直接用 JSON 更新页面上对应的 DOM 节点这套代码在监控大屏项目里已经被验证过多次const ws new WebSocket(ws://your-server/ws/ups/); ws.onmessage function (event) { const data JSON.parse(event.data); if (data.type ups_data_push) { document.getElementById(input-voltage).textContent data.input_voltage V; document.getElementById(battery-capacity).textContent data.battery_capacity %; } };5.4 心跳与断线重连机制WebSocket 长连接在公网或者复杂网络环境下很容易因为中间设备的 idle timeout 被悄悄断开。前端如果没发现连接已经断开就不会再收到任何数据更新页面看起来像卡住了一样。这个问题必须用心跳机制来解决。我在两个方向做了处理。服务端消费者里增加了一个周期任务每 30 秒往客户端发一条 ping 消息前端收到任何消息都会重置一个 watchdog 定时器如果超过 45 秒没有收到任何消息就判定连接已断开主动调用reconnect()重建连接。这个机制在实践里极大减少了“页面没数据但人没发现”的情况。再补充一个细节心跳消息要尽量轻量只带一个类型字段不要在里面塞数据快照否则浪费带宽。6. 双通道并行时的数据一致性与故障隔离6.1 同一份数据两条通道不能打架两套平台如果各自显示不同的电压值运维人员很可能最先怀疑 UPS 本身出了问题而实际上只是数据链路不一致。解决这个隐患要靠一个全局唯一的“数据快照 ID”。我在每次采集周期里生成一个自增的snapshot_id和时间戳这个快照同时进入通道 A 和通道 B。老平台侧虽然只关心 OID 值但我把snapshot_id映射到一组私有的 OID 里留给后续排查问题用WebSocket 推送则直接带上snapshot_id和collect_time字段。这样两套平台如果同时记录数据按时间戳对齐之后应该完全一致对不上账时可以快速定位是采集问题还是推送问题。6.2 通道故障的隔离与互不影响并行方案最大的优势就是故障隔离。我的代码结构里通道 A 和通道 B 是两个完全独立的模块各自拥有自己的异常捕获、重试机制和日志记录。通道 B 因为 Redis 临时不可用导致消息积压时通道 A 完全不受影响老平台照常读 OID反之如果老平台那边因为网络问题连不上 Agent通道 B 的 WebSocket 推送照常进行。这里有一个设计原则值得强调任何下游平台的异常都不能反向污染采集主流程。所以在采集进程和两个通道之间我用的是内存消息队列解耦采集进程只负责把快照写入队列通道 A 和通道 B 各自从队列里取数据发送。如果某个通道阻塞最坏结果只是它自己的队列积压主进程不会卡住。6.3 数据补发与消息幂等有些场景下通道断了之后数据是要补的。比如 WebSocket 断线两分钟期间发生了市电中断、电池放电这种关键事件前端看板重连后如果看不到这段时间的数据操作人员会漏掉重要信息。我在 WebSocket 消费者里加了历史快照补发逻辑前端重连成功时携带一个last_snapshot_id参数给服务端服务端查出缺失的快照列表并逐条补推。补推的时候要注意消息幂等。客户端收到补推数据后根据snapshot_id去重已经处理过的直接丢弃避免重复数据导致曲线跳变。7. 常见问题与排查技巧实录7.1 WebSocket 连接一建立就被断开这个问题我在项目初期反复遭遇。现象是浏览器控制台显示连接已建立但不到一秒钟就关闭。排查下来原因有三个按出现概率从高到低排第一没有用 Daphne 启动 ASGI 服务而是用了 Django 自带 runserver导致 WebSocket 握手直接失败。注意 Django 3.0 之后就算装了 Channels 也要显式指定 Daphne 来启动。第二Redis channel layer 版本不兼容新版channels-redis要求redis-py必须是 4.x 以上。第三反向代理层没有对 WebSocket 协议放行。7.2 SNMP OID 查询超时或者取不到数据这个问题通常是三个方向。一是队列里的 OID 写错了尤其是多相输入电压这种需要遍历子节点的 OID很多新手图省事直接查父节点结果返回超时。二是设备管理卡的 SNMP 服务没有启动或者 community 字符串和实际配置不一致。三是网络问题特别是跨网段访问时UDP 161 端口被防火墙拦截表现就是超时或者 timeout。我的排查习惯是先用命令手动验证一遍确认 OID 和 community 是否正常再去看代码。命令就一条snmpget -v2c -c public 192.168.1.20 1.3.6.1.2.1.33.1.4.1.0这条能通代码里大概率也没问题剩下就是代码层面的传参和超时设置了。如果命令都不通就换设备 IP、community、OID 三个变量逐个排查。7.3 两套平台同一个指标数值对不上这个问题的根源多半不在采集层而在下游平台各自的数据处理方式。老平台如果做了数据归档或者数据压缩存储的可能是五分钟均值而 Web 看板看到的是瞬时值两者自然会存在差异。还有一种情况是单位缩放不一致SNMP 里电压值是整数 2205老平台除以 10 变成 220.5V而新平台代码里忘了除以缩放系数直接显示 2205V一眼就能看出异常。解决方法是把缩放逻辑统一收敛到归一化模块里确保下游无论哪个通道拿到的都是真实物理值。数值一致性要有定期核对机制哪怕只是每天早上脚本对比一次两边的数据曲线。7.4 端口冲突与系统服务管理多通道并行导致系统里同时监听端口的服务明显变多。SNMP Agent 监听 UDP 161WebSocket 服务监听 TCP 8000Redis 监听 TCP 6379如果还开了 Modbus TCP 服务那监听端口就多了 502。每次服务启动都要确认端口没有被占我用ss -lntup | grep 端口号提前排查比 netstat 直观。另外两个推送通道和采集进程都要托管到 supervisor 里这样即使进程意外退出也能自动拉起并且每次异常退出都会留下日志方便事后定位问题。7.5 一条日志打通整条链路的思路双通道并行排查问题最怕的就是不知道问题出在哪一层。我在日志设计上做了一层约定每个环节打印标记前缀。采集层统一以[COLLECT]开头通道 A 以[CHANNEL-A]开头通道 B 以[CHANNEL-B]开头。日志内容只记录异常、状态切换、关键操作不记录正常数据帧否则日志量太大。举一个实际例子某天用户反馈看板数据卡住了但老平台正常。我第一反应去看[CHANNEL-B]日志发现 Redis 连接失败再看[COLLECT]日志完全正常问题定位在两分钟内完成。如果没有日志前缀靠人肉翻看混合日志效率会大打折扣。7.6 高频故障速查表为了方便现场运维我把这个项目里最高频的几个故障整理成了速查表放在文档里值班同事遇到问题可以直接对着表排查现象常见原因处理方式WebSocket 连接失败ASGI 服务未用 Daphne 启动用 daphne 启动命令如daphne -b 0.0.0.0 -p 8000 ups_monitor.asgi:applicationWebSocket 经常断线反向代理未配置 Upgrade 头、无心跳配置代理启动前端心跳重连SNMP 一直超时OID 错误、community 不对、UDP 161 被防火墙拦截先用 snmpget 验证再逐层排查Modbus 读取失败设备未开启 Modbus TCP、寄存器地址错误确认厂家手册的寄存器映射两平台曲线不一致单位换算、归档差值统一缩放逻辑、核对时间戳对齐采集进程反复崩溃超时未捕获异常、网络波动检查异常捕获加入 supervisord 自动重启写在最后这套双协议并行采集方案实际运行了两个月最直接的感受是把“数据模型”和“推送通道”彻底解耦是应对多平台对接最省心的做法。通道 A 是 SNMP 适配的 Agent通道 B 是 WebSocket 实时推送两者之间共用同一份数据快照互不依赖。后续如果需要接入第三套平台只要照着通道 A 或者通道 B 的模式写一个新适配器采集层一行代码都不需要改。最后分享一个调试时的小技巧刚开始跑双通道的时候不要一次性把两个通道都接上先把 WebSocket 通道跑通观察前端页面数据曲线是否平滑再切到通道 A单独验证老平台的轮询读取两边各自验证稳定之后再合到一起跑并行模式。这样能快速定位问题出在哪个环节而不是在两个通道同时运转的时候被各种报错搞得手忙脚乱。