简介这是一份基于C实现的轻量级Modbus TCP服务器开源项目面向工业自动化开发者、嵌入式系统工程师及学习网络协议编程的中高级学习者用于快速搭建符合Modbus TCP规范的设备通信服务端。资源包共14个文件含5个头文件h定义协议结构与核心类接口5个源文件cpp实现TCP监听、报文解析、寄存器读写及错误响应等关键逻辑另有Visual Studio解决方案sln、工程配置vcproj、日志模块Log.h/cpp、全局配置global.h和协议封装modbus.h/cpp等结构清晰、模块职责分明压缩包仅13KB便于快速导入与二次开发。已有680人学习下载读者可直接复用其多线程TCP服务框架、标准Modbus功能码处理流程及寄存器映射机制深入理解工业协议在以太网环境下的C落地实践并基于README.md和示例代码快速启动调试。1. Modbus TCP服务器不是“接上线就能通”的黑匣子它得先听懂工业现场的节奏再扛住PLC轮询风暴Modbus TCP服务器这个标题里藏着三重现实张力第一层是协议——它不是HTTP那种带状态、有重试、能自愈的现代协议而是裸TCP功能码寄存器地址的极简组合第二层是场景——它跑在工厂产线边缘设备上面对的是西门子S7-1200、三菱FX5U这类PLC每200ms一次的稳定轮询不是浏览器偶尔点一下的请求第三层是落地陷阱——很多人用Python写个socket.listen()就以为成了服务器结果一接真实PLC立刻出现超时、乱码、寄存器值跳变、甚至整个连接被复位。这不是代码写错了而是没理解Modbus TCP的本质它是一套带时序约束的寄存器映射协议服务器端必须严格遵循“请求→解析→查表→组包→响应”这一毫秒级闭环且对异常如非法地址、功能码不支持的响应格式比正常响应更关键。本文面向两类人一是刚从Web后端转嵌入式通信的开发者需要把HTTP思维清零二是自动化集成工程师手头有现成PLC但缺一个可调试、可监控、可嵌入的轻量级服务端。我们不讲OSI七层模型只讲怎么让第一个读线圈0x01请求在Wireshark里看到干净的0x01响应包以及为什么第17次轮询时你的Python服务器突然卡住3秒——那不是GC是寄存器锁没释放。2. 从零手写一个可调试的Modbus TCP服务器用pymodbus搭骨架再亲手补上工业级心跳与寄存器快照2.1 为什么不用现成的“一键启动”服务——协议栈分层决定你必须控制最底层市面上有pymodbus、libmodbus、modbuspal等成熟库但直接调StartTcpServer()会埋下三个隐患第一pymodbus默认使用AsyncioModbusTcpServer其事件循环与业务逻辑耦合过深一旦你在custom_handler里加了串口读取或数据库写入整个TCP响应就会被阻塞第二它的寄存器存储是纯内存字典没有原子性保护当多个PLC并发读写同一地址段时会出现值覆盖第三它不提供请求级日志钩子你无法知道某次0x16写多个寄存器失败是因为地址越界还是CRC校验未启用Modbus TCP其实不用CRC但很多PLC仍会校验报文长度。所以我们选择pymodbus作为协议解析层它对MBAP头、功能码、数据域的拆解极其精准但自己实现存储层和调度层。这不是重复造轮子而是把“协议合规性”和“业务可靠性”解耦——前者交给pymodbus验证后者由你掌控。2.2 最小可运行服务器12行核心代码跑通读线圈0x01流程from pymodbus.server import StartTcpServer from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext from pymodbus.transaction import ModbusSocketFramer import asyncio # 1. 初始化4个寄存器区线圈(0x0000)、离散输入(0x1000)、输入寄存器(0x3000)、保持寄存器(0x4000) store ModbusSlaveContext( diModbusSequentialDataBlock(0, [0]*100), # 离散输入只读 coModbusSequentialDataBlock(0, [0]*100), # 线圈可读写 hrModbusSequentialDataBlock(0, [0]*100), # 保持寄存器可读写 irModbusSequentialDataBlock(0, [0]*100), # 输入寄存器只读 ) context ModbusServerContext(slaves{0x01: store}, singleTrue) # 2. 启动服务器注意这里用同步模式避免async干扰调试 if __name__ __main__: StartTcpServer( context, address(0.0.0.0, 502), framerModbusSocketFramer )提示这段代码能让你用Modbus Poll工具连上并读取线圈0-9。但注意singleTrue参数——它强制所有请求都路由到slave ID0x01绕过多从站寻址逻辑适合单设备调试。真实产线中PLC可能发slave ID0x02的请求此时需设singleFalse并确保context中存在对应键。2.3 工业级增强给寄存器加“快照锁”防并发写崩数据一致性PLC轮询时常有多个请求并发到达比如请求0x03读保持寄存器和0x10写多个保持寄存器同时命中地址40001-40010。pymodbus默认的ModbusSequentialDataBlock内部用普通list无锁操作。我们替换为带读写锁的自定义块import threading from pymodbus.datastore import ModbusSparseDataBlock class ThreadSafeDataBlock(ModbusSparseDataBlock): def __init__(self, valuesNone, **kwargs): super().__init__(values, **kwargs) self._lock threading.RLock() # 可重入锁避免同一线程多次acquire死锁 def setValues(self, address, values): with self._lock: super().setValues(address, values) def getValues(self, address, count): with self._lock: return super().getValues(address, count) # 使用方式替换store中的hr字段 store ModbusSlaveContext( hrThreadSafeDataBlock({i: 0 for i in range(100)}), # 地址映射改为字典更易debug # 其他区域保持不变... )逻辑说明RLock可重入锁比Lock更适合此场景——因为pymodbus内部可能在setValues中再次调用getValues如做范围校验普通锁会导致死锁。参数说明{i: 0 for i in range(100)}生成地址0~99的初始值字典比[0]*100更直观调试时可直接print(store.hr._data)看当前值。2.4 心跳保活机制为什么PLC断连后你的服务器不报错——TCP Keepalive必须显式开启Modbus TCP本质是长连接但Linux内核默认TCP Keepalive时间是2小时而PLC通常30秒无数据就主动断连。若服务器不探测会一直维持着“假连接”直到下次发包才触发RST。解决方案是在socket层开启Keepalivefrom pymodbus.transport import ModbusTcpServer # 替换StartTcpServer手动构建server以注入socket选项 class CustomTcpServer(ModbusTcpServer): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 在socket创建后立即设置keepalive self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 30) # 首次探测前空闲秒数 self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10) # 探测间隔 self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3) # 失败次数阈值 # 启动时用CustomTcpServer替代 server CustomTcpServer(context, address(0.0.0.0, 502)) server.serve_forever()参数说明TCP_KEEPIDLE30表示连接空闲30秒后开始探测TCP_KEEPINTVL10表示每次探测失败后隔10秒再探TCP_KEEPCNT3表示连续3次失败则关闭连接。这三个值要匹配PLC侧配置否则PLC可能先断连服务器还傻等。3. 请求级日志与异常捕获看清PLC到底在发什么而不是猜“为什么不通”3.1 在协议解析前插桩打印原始MBAP头和功能码定位物理层问题很多“连不上”其实是网线没插牢或IP配错但错误日志只显示ConnectionRefusedError。我们在pymodbus请求处理链最前端加一层原始数据捕获from pymodbus.factory import ServerDecoder from pymodbus.utilities import hexlify_packets class LoggingRequestHandler: def __init__(self, handler): self.handler handler def __call__(self, request, *args, **kwargs): # 打印原始字节流MBAP头占7字节事务ID/协议ID/长度/单元ID if len(request) 7: tid int.from_bytes(request[0:2], big) pid int.from_bytes(request[2:4], big) length int.from_bytes(request[4:6], big) uid request[6] func_code request[7] if len(request) 7 else N/A print(f[DEBUG] MBAP: TID{tid} PID{pid} LEN{length} UID{uid} FUNC{func_code} RAW{hexlify_packets(request[:16])}) return self.handler(request, *args, **kwargs) # 注入到pymodbus的decoder中 decoder ServerDecoder() original_execute decoder.execute decoder.execute LoggingRequestHandler(original_execute)现象价值当PLC发来00 01 00 00 00 06 01 01 00 00 00 01时你能在日志看到FUNC1确认是读线圈若看到00 01 00 00 00 06 01 FF ...则FUNC255说明PLC配置了错误的功能码0xFF非法问题出在PLC程序而非服务器。3.2 拦截并重写异常响应让PLC明确知道“地址越界”而不是静默丢包Modbus标准规定当请求非法地址时服务器必须返回异常响应功能码0x80且异常码为0x02非法数据地址。但pymodbus默认行为是抛出ModbusIOException导致连接中断。我们重写异常处理器from pymodbus.exceptions import NotImplementedException, ParameterException from pymodbus.pdu import ExceptionResponse def custom_exception_handler(self, request, error): 自定义异常响应生成器 if isinstance(error, ParameterException): # 地址越界返回异常码0x02 return ExceptionResponse(request.function_code, 0x02) elif isinstance(error, NotImplementedException): # 功能码不支持返回0x01 return ExceptionResponse(request.function_code, 0x01) else: # 其他错误统一返回0x04从站设备故障 return ExceptionResponse(request.function_code, 0x04) # 注入到context context.error_handler custom_exception_handler注意此函数必须绑定到context对象而非server实例。pymodbus在execute阶段会检查context.error_handler是否存在存在则调用它否则走默认逻辑。3.3 实时寄存器监控用HTTP接口暴露当前值告别“改完代码重启看效果”调试时最痛苦的是改了线圈0的值却不知PLC是否真的读到了。我们加一个轻量HTTP端点用Flask不引入新框架from flask import Flask, jsonify, request import threading app Flask(__name__) app.route(/registers/string:area/int:address, methods[GET, POST]) def handle_register(area, address): if area not in [co, hr, ir, di]: return jsonify({error: invalid area}), 400 block getattr(store, area) if request.method GET: value block.getValues(address, 1)[0] return jsonify({address: address, value: value}) else: # POST data request.get_json() block.setValues(address, [data[value]]) return jsonify({status: updated}) # 启动HTTP服务在独立线程避免阻塞Modbus主线程 threading.Thread(targetlambda: app.run(host0.0.0.0, port5000, debugFalse)).start()使用方式curl http://localhost:5000/registers/hr/0查看保持寄存器0的值curl -X POST http://localhost:5000/registers/co/0 -H Content-Type: application/json -d {value:1}强制置位线圈0。这比反复启停Modbus服务高效十倍。4. 避坑PLC轮询场景下5个血泪经验总结第3条90%的人第一次都踩4.1 现象PLC连接后能读取但2分钟后自动断开Wireshark显示FIN包原因PLC侧设置了30秒无数据超时而你的服务器未发送任何心跳Modbus无心跳帧依赖TCP Keepalive。pymodbus默认不开启socket keepalive且Linux内核keepalive默认2小时。解决按2.4节代码显式设置TCP_KEEPIDLE30并与PLC手册确认其超时值取最小值。4.2 现象用Modbus Poll工具读取正常但西门子PLC报“连接超时”原因PLC严格校验MBAP头中的协议IDProtocol Identifier标准值为0x0000但某些pymodbus版本在构造响应时可能填0x0001。解决抓包对比——用Wireshark过滤modbus tcp.dstport 502看响应包MBAP头第2-3字节是否为00 00。若不是在pymodbus/pdu.py中搜索protocol_id强制设为0。4.3 现象多个PLC同时连接读取保持寄存器时值随机跳变原因ModbusSequentialDataBlock的getValues方法内部用list[slice]而Python list切片不是原子操作。当A线程正在getValues(0,10)B线程执行setValues(5, [1])可能读到部分旧值部分新值。解决必须用3.3节的ThreadSafeDataBlock且锁粒度要覆盖整个getValues和setValues调用不能只锁内部赋值。4.4 现象写多个寄存器0x10成功但PLC侧实际只更新了前2个原因Modbus TCP要求请求数据长度必须是偶数字节因寄存器为16位而pymodbus对setValues传入奇数长度列表时会自动补0但PLC可能拒绝这种“非标准填充”。解决在调用setValues前校验长度if len(values) % 2 ! 0: raise ValueError(values length must be even)并在PLC侧确保写入数量为偶数。4.5 现象服务器CPU飙升至100%top显示python进程占满一个核原因pymodbus的StartTcpServer在singleFalse模式下会为每个slave ID创建独立线程若PLC误发slave ID0xFFpymodbus会不断创建新线程处理无效请求最终耗尽资源。解决启用singleTrue单从站模式或在context初始化时预设所有可能的slave ID如{0x01:store, 0x02:store}避免动态创建。5. 进阶技巧用寄存器快照做PLC指令回放把“线上问题”变成“本地可复现Case”5.1 为什么需要指令回放——产线问题无法截图只能靠“当时PLC在读什么”工厂现场最头疼的场景PLC某天凌晨3点突然停止运行日志只有一行“Modbus timeout”但服务器日志无异常。你无法复现因为问题可能和温度、电压波动、PLC固件bug相关。此时唯一可靠证据是那个时刻PLC发出的所有请求。我们设计一个环形缓冲区记录最近1000次请求的原始字节、时间戳、客户端IPfrom collections import deque import time class RequestRecorder: def __init__(self, max_size1000): self.buffer deque(maxlenmax_size) self.lock threading.Lock() def record(self, client_ip, raw_request): with self.lock: self.buffer.append({ timestamp: time.time(), ip: client_ip, raw: raw_request.hex(), # 存储十六进制字符串节省内存 size: len(raw_request) }) recorder RequestRecorder() # 注入到请求处理前在LoggingRequestHandler中调用 def log_and_record(client_ip, request): recorder.record(client_ip, request) # ... 原有日志逻辑5.2 从快照生成可复现的测试Case用pymodbus的ModbusClient重放请求当问题发生后你从recorder.buffer中取出一个异常请求如00 01 00 00 00 06 01 03 00 00 00 02保存为case_20240501_030001.bin。然后写一个重放脚本from pymodbus.client import ModbusTcpClient from pymodbus.transaction import ModbusSocketFramer import sys def replay_case(case_file): with open(case_file, rb) as f: raw f.read() client ModbusTcpClient(127.0.0.1, port502, framerModbusSocketFramer) client.connect() # 直接发送原始字节绕过pymodbus编码 client.socket.send(raw) response client.socket.recv(1024) print(fRequest: {raw.hex()}) print(fResponse: {response.hex()}) client.close() if __name__ __main__: replay_case(sys.argv[1])提示此脚本不经过pymodbus的request构造直接socket发送因此能100%复现PLC当时的二进制行为。配合Wireshark抓包你能对比“PLC发的”和“你的服务器回的”精准定位是解析错、还是响应慢、还是寄存器值不对。5.3 寄存器变更追踪当PLC写入时自动触发业务逻辑如告警、存库很多项目需要“PLC写入D100时发邮件通知”。pymodbus不提供写入钩子但我们可以在ThreadSafeDataBlock.setValues中注入回调class CallbackDataBlock(ThreadSafeDataBlock): def __init__(self, callbacksNone, **kwargs): super().__init__(**kwargs) self.callbacks callbacks or {} def setValues(self, address, values): super().setValues(address, values) # 触发注册的回调 for (cb_addr, cb_func) in self.callbacks.items(): if cb_addr address or (isinstance(cb_addr, tuple) and cb_addr[0] address cb_addr[1]): cb_func(address, values) # 使用监听地址0-9的写入 def on_co_write(addr, vals): print(fCoil {addr} written with {vals}) # 这里加邮件发送、数据库写入等业务逻辑 store ModbusSlaveContext( coCallbackDataBlock( callbacks{0: on_co_write}, # 地址0写入时触发 values[0]*100 ), # 其他区域... )参数说明callbacks字典的key可以是整数精确地址或元组地址范围value是回调函数。这样业务逻辑与协议栈完全解耦增删功能只需改callback字典无需碰Modbus核心代码。我做过的最深的一个坑是某次产线升级PLC固件后新固件在写入保持寄存器时会额外发送一个0x17报告从站ID请求。我们的服务器没实现0x17按标准应返回异常但pymodbus默认忽略未知功能码导致PLC认为“指令未送达”反复重发最终挤爆网络。后来我在custom_exception_handler里加上对0x17的显式拦截问题立解。这件事让我明白Modbus服务器不是“支持多少功能码”而是“对每一个PLC可能发来的字节都有确定的、符合标准的响应”。希望帮到你。本文还有配套的精品资源点击获取