I/O多路复用实战：用selectors重构Modbus TCP多从站采集，性能提升8倍

去年有个项目把我逼到墙角现场一台工控机要采集32台变频器的Modbus TCP数据加上车间里20多个温度采集模块林林总总五十多个从站。刚开始交付时用最常规的串行轮询代码是好写可跑起来数据刷新周期一路飙到三秒多调度屏上的曲线全是锯齿。用户拍着桌子问能不能做到一秒以内我第一反应是上线程池后来才真正想明白问题不在CPU不够快而在网络等待被白白浪费掉了。那天晚上我重新翻了一遍I/O多路复用的东西用selectors重写了采集层数据刷新周期从3秒压到了400毫秒。这篇就把整个思路、报文解析、代码骨架和现场踩过的坑一次性说透。适合正在用Modbus TCP攒上位机、遇到多从站采集性能瓶颈、或者考虑用I/O多路复用替换轮询方案的工程师。看完可以直接参考这套结构去改写自己的采集程序。1. 轮询不是不行是挂多了真不行1.1 轮询是怎么一步步拖垮采集周期的串行轮询的逻辑非常直白建立一个socket列表循环给每个从站发请求发出一个就等一个响应拿到数据解析完再去问下一个。伪代码大概是这个样子for device in devices: frame build_modbus_tcp_request(device) sock.sendall(frame) response wait_response(sock, timeout1.0) if response: parse_response(response)这套逻辑在从站数量少的时候完全没问题。现场只有一两台仪表一秒钟轮几十次都可以。可一旦从站数量超过20个问题就暴露了每个请求都要等一个完整的网络往返时间。假设一个从站从发出请求到收到响应需要30毫秒50个从站就是1500毫秒这还没算上应用层的解析时间和可能出现的超时等待。更糟糕的是串行模式下某个从站响应慢、或者干脆丢包了它后面排队的从站全都要跟着等超时。我最早的项目就是这种经典写法从站少的时候没人说什么等点位表扩充到一千多点调度屏刷新就开始肉眼可见地卡顿。后来加过各种“优化”把超时从1秒改成200毫秒把多个寄存器的读取数量加大把读取频率降下来。这些都是治标不治本因为你始终在串行等待网络I/O没有得到任何重叠。1.2 多线程为什么也不是银弹很多人的第二反应是上多线程给每个从站开一个独立线程各等各的响应总时间不就等于最慢的那台设备了吗理论上确实如此但实际工程里坑并不少。第一线程数量会跟着从站数量线性膨胀。管50个从站就要维护50个线程每个线程里还有独立的socket、超时循环、重连逻辑代码稍微一复杂就很容易出现资源没释放、socket泄漏的问题。第二Python这类语言还有GIL线程一旦从网络等待状态切换到数据处理锁竞争的问题就会浮出水面。即使你用C#或Java线程的上下文切换开销和内存占用也不是白给的。第三多线程方案里你依然要自己处理“哪些请求已经发出去、哪些响应还没回来”的账本账本一乱对应的响应就不知道扔给谁。本质上多从站高速通讯的瓶颈不是计算而是网络等待。I/O多路复用正是干这件事的用一个线程同时盯着所有socket哪个socket有数据可读就去处理哪个。Linux上是epollWindows上selectors会自动选择合适的机制。这也是“多路复用”这个词在Modbus TCP场景里的真正含义——不是把多个串口信号合到一起而是把多个网络连接的就绪事件合并到一个事件循环里统一处理。2. 多路复用前先把Modbus TCP报文的“身份证”搞明白2.1 MBAP头里藏着事务ID和单元ID要在多路复用的场景里正确处理Modbus TCP必须把报文结构刻在脑子里。Modbus TCP帧分两部分MBAP头7字节和PDU。PDU就是功能码加数据真正干活的部分MBAP头则是给TCP传输用的“信封”。MBAP头的四个字段字段长度说明事务标识符2字节客户端生成每次请求自增响应里原样返回协议标识符2字节固定为0表示Modbus协议长度2字节表示后面还有多少个字节等于PDU长度加1单元标识符1字节对应Modbus从站地址相当于报文的“收件人”事务标识符这个东西在串行轮询里几乎用不到因为一个请求对应一个响应顺序不会乱。但是在多路复用下你一口气给同一个socket发了10个请求每个请求的响应时间可能不同响应返回的顺序也未必和请求顺序一致。这时候就必须靠事务标识符来匹配“这个响应对应的是哪个请求”。说得直白点事务ID就是每封请求信的编号谁的回信就贴谁的编号。举个例子读1号从站保持寄存器起始地址0读10个寄存器完整的请求帧是这样00 01 00 00 00 06 01 03 00 00 00 0A拆开看00 01是事务ID00 00是协议ID00 06是后续长度01是单元ID03是功能码读保持寄存器00 00是起始地址00 0A是寄存器数量。响应帧会把你发出去的事务ID原样带回来后面的长度字段是2寄存器数×2再往后是数据字节数加寄存器值。这里有一个经常被忽略的点如果设备返回的功能码最高位是1比如0x83说明它返回的是一个Modbus异常响应跟在后面的一个字节是异常码01非法功能、02非法数据地址、03非法数据值、04从站设备故障。解析的时候必须先判断功能码再决定按正常数据帧解析还是按异常帧处理否则很容易把错位的数据当成真实数值写进数据库。2.2 TCP长连接为什么比短连接更适合多从站搜索热词里常年出现“tcp三次握手四次挥手”“tcp长连接与短连接”这恰恰是Modbus TCP主站设计里容易想当然的地方。如果每个请求都新建一个TCP连接那么每次请求都要先完成三次握手结束后再四次挥手。假如同一个机房里的设备RTT是5毫秒一次握手加挥手就额外消耗掉十几个毫秒高频采集下这个损耗相当可观。所以多从站高速通讯必须使用长连接从站设备一上电主站就把socket建好后续所有请求都在这条连接上复用。长连接带来的另一个好处是你可以使用流水线方式在一个连接上连续发出多个请求不用等一个响应回来再发下一个。这个能力是压榨吞吐量的核心。但长连接也有代价一旦网络断开但双方没有及时感知就会出现半开连接socket看着还在实际上数据已经发不出去了。后面第5节会专门聊怎么处理掉线和半开连接。另外一些PLC自带的Modbus TCP Server对同时建立的连接数有硬限制连接池如果不做上限和复用很容易把设备连接数打满。3. 单线程搞定多从站一个可跑的I/O多路复用主站3.1 整体结构连接池加事件循环我最后定下来的架构不复杂三个部分组成连接池每个从站对应一个非阻塞socket启动时统一建立连接。事件循环所有socket注册到selectors.DefaultSelector主循环里等待就绪事件谁有响应就处理谁。请求窗口每个从站连接上同时允许一定数量的未确认请求超出就等后续再发防止把嵌入式设备压垮。为什么要把“连接池”单独拿出来因为如果一个从站IP下有多个单元ID或者某个网关设备承担多个从站你不能无脑给每个从站都建物理连接。正确做法是先按“物理连接”划分再在一条连接上按照unit_id区分不同从站。同一个socket上的多个请求就靠事务ID和unit_id共同确定响应归属。请求窗口也很关键。Modbus TCP协议本身没有禁止在一个连接上同时发多个未确认请求但这不代表所有设备都扛得住。很多芯片级Modbus从站只能一个请求一个请求地处理你一次丢进去10个请求响应还是会按顺序排着回来甚至有的设备会直接丢掉来不及处理的请求。所以窗口大小需要做成可配置项常见设备可以从4到8开始试。3.2 核心代码selectors版Modbus TCP主站我贴一段自己工程里简化后的核心骨架用Python写重点看事件处理和报文匹配的逻辑。import socket import selectors import struct import time sel selectors.DefaultSelector() class ModbusTcpClient: def __init__(self, host, port502, unit_id1): self.host host self.port port self.unit_id unit_id self.sock None self.buffer b self.tid 0 self.pending {} self.on_ready None def connect(self): self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setblocking(False) try: self.sock.connect((self.host, self.port)) except BlockingIOError: pass sel.register(self.sock, selectors.EVENT_READ | selectors.EVENT_WRITE, self) def handle_event(self, mask): if mask selectors.EVENT_WRITE: # 连接建立完成切到只读事件触发外部回调 sel.modify(self.sock, selectors.EVENT_READ, self) if self.on_ready: self.on_ready(self) if mask selectors.EVENT_READ: self._recv() def build_request(self, function_code, start_addr, quantity): self.tid (self.tid 1) 0xFFFF pdu struct.pack(BHH, function_code, start_addr, quantity) mbap struct.pack(HHHB, self.tid, 0, len(pdu) 1, self.unit_id) return self.tid, mbap pdu def send_request(self, function_code, start_addr, quantity, callback): tid, frame self.build_request(function_code, start_addr, quantity) try: self.sock.sendall(frame) except (BrokenPipeError, ConnectionResetError): self.close() return False self.pending[tid] (callback, time.monotonic()) return True def _recv(self): try: data self.sock.recv(4096) except BlockingIOError: return if not data: self.close() return self.buffer data self._parse_frames() def _parse_frames(self): while len(self.buffer) 6: tid, proto, length struct.unpack(HHH, self.buffer[:6]) if len(self.buffer) 6 length: return frame self.buffer[:6 length] self.buffer self.buffer[6 length:] self._dispatch(tid, frame) def _dispatch(self, tid, frame): entry self.pending.pop(tid, None) if entry: entry[0](frame) def check_timeout(self, timeout3.0): now time.monotonic() for tid in list(self.pending.keys()): _, sent_at self.pending[tid] if now - sent_at timeout: self.pending.pop(tid) print(f{self.host} request timeout, tid{tid}) def close(self): try: sel.unregister(self.sock) except Exception: pass self.sock.close()主循环的调度部分是这样clients [] for cfg in from_config(slaves.json): client ModbusTcpClient(cfg[host], cfg[port], cfg[unit_id]) client.on_ready lambda c: c.send_request(3, 0, 10, parse_response) clients.append(client) for c in clients: c.connect() running True while running: events sel.select(timeout0.1) for key, mask in events: key.data.handle_event(mask) for c in clients: c.check_timeout(timeout3.0)这段代码的核心在于事件循环先触发所有连接就绪事件然后每个连接在on_ready里发出自己的第一轮请求。之后内核会在任意一个socket收到数据时唤醒select_recv只负责把数据追加到缓冲区_parse_frames则根据MBAP头的长度字段把完整帧切出来最后按事务ID找到对应的回调。注意_parse_frames里的循环只依赖长度字段不依赖“一次recv恰好收到一帧”的假设。TCP是字节流一次recv可能收到几个响应也可能只收到半个响应这个循环能同时处理粘包和半包。3.3 为什么“同时发出去”效果显著你可以把串行轮询想象成去银行柜台办业务一个窗口只有一个柜员后面的人必须等前面的人办完。I/O多路复用则像是把一堆业务单同时递进窗口柜员处理完哪个就叫哪个号客户的等待时间重叠在了一起。从灵魂深处讲I/O多路复用并没有让单次Modbus请求变快它只是让多个socket的等待时间重叠。主站在一个循环里把所有请求都丢进TCP发送缓冲区网络协议栈会自己处理并发传输哪个连接的响应先到达应用层就去解析哪个。最终一轮采集的耗时从“所有从站耗时之和”变成了“最慢那个从站的耗时”这就是吞吐量能提升几倍到十几倍的原因。4. 不同方案实测对比轮询、线程池和事件驱动的差距4.1 测试环境与模拟方法纸上谈兵没意思我把自己写的主站程序和两种传统方案放在同一台机器上做了对比。测试环境是i5工控机Linux系统从站侧用Docker起了32个模拟Modbus TCP从站。每个从站都模拟100个保持寄存器响应时间人为加了5到20毫秒的随机延迟模拟真实工业现场的抖动。采集任务统一为读取32个从站每个地址0到9的10个寄存器。三种方案分别是串行阻塞socket一个循环发一个请求等一个响应。线程池32个线程每个线程独立socket阻塞读取。I/O多路复用也就是上文那套selectors实现请求窗口限制为8。4.2 数据说话跑下来的数据大概是这样方案单轮采集耗时CPU占用代码复杂度串行阻塞1.8秒左右低最低32线程0.35秒左右中中等I/O多路复用0.35秒左右低偏高线程池和I/O多路复用的总耗时都在350毫秒附近因为瓶颈已经是网络RTT和模拟从站的响应时间而不是主站并行能力。真正的差别在CPU占用和扩展性上。32线程方案在Windows上内存和句柄数都会涨线程切换也不是零成本用I/O多路复用即使从站数翻到100个线程数仍然是1资源占用几乎不变。如果从站响应时间差异巨大多路复用的优势会更明显。比如从站A响应5毫秒从站B响应300毫秒串行轮询的总耗时会接近“所有从站响应时间之和”多路复用则只由最慢的那台决定。现场经常出现一两台设备因为布线原因响应很慢多路复用方案能有效把这些慢设备的影响范围降下来。4.3 什么时候多路复用反而没意义这里必须泼一盆冷水如果从站不是真正的以太网设备而是通过“Modbus RTU over TCP网关”挂在485总线上那多路复用基本没有意义。因为网关背后是串口总线同一时刻只能有一个请求在总线上跑客户端并发发出去100个请求网关也是排着队一个个处理。并发不仅不能提升速度反而可能导致请求堆积、缓冲区溢出甚至让网关崩溃。判断方法很简单看每个从站的IP和端口。如果32个从站是32个不同的IP地址大概率是纯以太网设备可以并发如果32个从站共用一个IP端口都是502那多半是串口服务器或者网关这类设备老老实实串行轮询或者只在应用层做并发把请求发出去但心里要清楚瓶颈在网关内部。我见过有人对着一个16路串口服务器疯狂并发结果网关频繁死机最后换回串行轮询反而稳定。5. 现场最容易翻车的几个细节5.1 粘包半包长度字段是拆包的地图多路复用主站最典型的bug就是报文拆错。TCP协议是字节流没有应用层的消息边界一次recv可能把两个响应一起读进来也可能只读到一个响应的前半截。如果按“一次recv等于一个响应”来解析轻则数据错位重则整个解析流程崩溃。正确做法就是上文代码里的思路先把数据追加到一个缓冲区然后循环检查缓冲区头部7字节里的长度字段长度字段告诉你“这一帧完整大小是多少”缓冲区不够就继续等下一次recv够了就切出一帧剩下的留到下一轮继续切。这个逻辑对所有基于Modbus TCP的采集程序都适用不管你用不用I/O多路复用。另一个隐藏问题是重复响应。TCP超时重传、应用层重试、从站设备逻辑异常都有可能导致主站收到两个事务ID相同的响应。处理方式是在_dispatch里用pop而不是get取回调这样第一个响应处理完第二个响应就只能被丢弃不会进入解析流程。5.2 超时与掉线最怕卡住不动现场网络不可能永远稳定。一个从站掉线、一个交换机端口松动、一根网线接触不良都是再常见不过的事。对多路复用主站来说最怕的并不是某个socket出错而是某个socket“不死不活”——连接还在但永远没有数据回来请求发出去就石沉大海。所以每个pending请求都必须记录发出时间事件循环里定期扫描超过阈值直接丢弃。我在代码里用的是check_timeout每0.1秒查一次超时默认3秒实际项目里可以根据点位刷新要求调整。注意超时时间不能太短否则正常响应慢一点都被误判成超时也不能太长否则用户看数据的时候会感觉“卡死”。掉线重连不能一上来就把socket断开。连续丢几个超时再认为设备掉线然后关闭socket并从selectors注销。之后按退避策略重连从1秒开始失败次数多了逐步加到10秒同时加一点随机抖动避免32个从站同时掉线重连时把网络打满。5.3 从站连接数限制与unit id复用有些PLC的Modbus TCP Server对并发连接数限制非常死。之前遇到过一台S7-1200作为服务端最多允许几个客户端连接一旦超过后来的连接直接被拒。所以主站侧必须做连接池不能每个请求都新建socket也不能无限制地给每个从站分配独立连接。如果设备支持一个IP带多个单元ID尽量在一条TCP连接上复用。请求报文里的unit_id字段就是干这个用的。但还是要强调Modbus TCP规范允许同一连接上多个未确认请求实际设备不一定按照这个理想模型工作。有的设备无论你发得多快它都响应完一个再处理下一个。这时候把单连接上的请求窗口调小比如固定为1退化成“半串行”模式但依然能省去TCP重连的开销。5.4 502端口和防火墙Modbus TCP默认端口是502很多设备配置的时候也允许改成自定义端口。现场最常见的问题是防火墙把端口挡了从站明明在线主站就是连不上。Windows防火墙、Linux的firewalld或者iptables都要检查一遍。尤其是有的现场工控机会装安全软件悄无声息地把502端口拦掉排查半天才发现是这个原因。另一个容易忽略的细节是如果用了NAT或者多网卡主站程序监听的本地端口和从站通信的网口路由也要确认。双网卡工控机很常见一个网口接办公网一个网口接工业网程序必须绑定正确的网卡地址否则socket connect会走默认路由到错误的网段。6. 从Demo到采控服务稳定性才是硬功夫6.1 连接生命周期管理前面的代码骨架可以跑通基本采集但真正上线还差不少工程化工作。连接生命周期要分状态管理初始化、连接中、在线、超时、重连等待、关闭。建议用一个状态机别用一堆if-else散落在代码里。重连策略单独强调一点不要在事件循环里同步等待重连。非阻塞socket的connect可以立即返回事件循环继续跑其他socket重连只注册一个EVENT_WRITE事件等待内核通知连接是否成功。这样某台设备掉线的时候其他设备的采集完全不受影响。如果某个设备反复掉线要把它从常规采集队列里摘出来单独降级处理别让它每次超时都拖慢一轮整体扫描。6.2 数据一致性与采集线程解耦事件循环是单线程的任何阻塞操作都会阻塞所有socket的收发。所以数据库写入、UI刷新、日志落盘这些操作绝对不能直接写在回调函数里。我的做法是采集回调里只更新内存里的“最新值缓存”外部再统一由定时任务把缓存批量落库。这样即使数据库偶尔变慢也不会影响实时采集。缓存结构可以是一个字典key是(ip, unit_id, 寄存器地址)value是最新值加时间戳。展示层读取的时候直接拿这个字典查询线程只需要加一把锁保护写入读取的并发安全。用双缓冲或者原子引用也可以核心原则是事件循环不碰任何可能阻塞的东西。6.3 上线前用Modbus Slave和Wireshark交叉验证我开发阶段最喜欢用的两个工具是Modbus Slave和Wireshark。Modbus Slave可以模拟从站把自己写的请求丢给它看返回帧的结构是否和预期一致Modbus Poll反过来当主站用来确认设备本身的通信能力。多路复用代码写完以后先别急着接现场设备。用Modbus Slave模拟十几个从站每个从站设不同的unit_id打开Wireshark抓包重点确认三件事事务ID是否正确随请求递增、同一TCP连接上是否真的出现多个未确认请求、响应返回时对应的事务ID能否和请求匹配上。我见过有同事在代码里忘了一次事务ID自增结果所有请求ID都一样响应一回来全部被当成重复包丢掉现场排查了两个小时。6.4 日志里必须有事务生命周期多从站并发最怕出问题说不清楚是谁的锅。所有请求和响应的生命周期都应该能通过日志追踪。我习惯每条请求打一行结构化日志时间、从站IP、unit_id、功能码、起始地址、寄存器数量、事务ID、耗时、返回状态。超时和异常响应额外打WARN级日志。这样现场用户报“某个点读数不对”翻日志就能定位是主站没发出请求还是从站返回了异常码还是响应在应用层被解析错了。最后说句实在话。I/O多路复用不是银弹拿到项目先分清是纯以太网多从站还是串口网关透传是4台设备还是100台设备设备的Modbus实现是否支持并发请求。搞清楚这些再选方案。如果只是三五台PLC轮询串行代码反而省事但一旦点位规模上来这套事件驱动的采集架构值得你花两天时间重构一遍。后面还能把同一套事件循环扩展成同时采集Modbus TCP和Modbus RTU再往上层接MQTT转发整个采集层就能统一管理起来不用维护好几套代码了。