简介:面向工业互联网环境的数据采集与设备调试需求,这款Modbus TCP主从测试软件支持Master与Slave两种工作模式。作为主站时,可发起Modbus请求并指定功能码,如读取输入寄存器、写入单个线圈,用于检验从站设备的响应与数据完整性;切换至从站模式,则模拟设备端接收主站命令,方便开发人员调试服务器端应用。软件提供实时寄存器数据查看与定时采集记录功能,能直观监控设备状态变化,辅助故障定位与性能优化。资源包共97个文件、2.2MB,以C#工程源码(cs、csproj、sln)为主体,包含exe程序、config配置、dll库及txt说明等,结构清晰,便于快速编译与二次开发。由于源码完整开放,使用者可深入理解Modbus TCP协议实现细节,按需扩展功能或集成至现有系统。该工具目前已吸引2022人学习使用,适合工业自动化工程师、PLC调试人员及协议开发者在开发、测试和维护Modbus TCP设备时直接采用。
1. modbusTCP Master/Slave测试软件:现场验证中最常被用来摸设备底细的工具
搞过现场调试的人都明白,设备有没有配好地址、寄存器里到底是什么数据类型、响应速度能不能跟上系统轮询,这些问题在没有modbusTCP Master/Slave测试软件之前,全靠一遍遍改主站代码和人肉猜。这个测试软件就是一个能随时切换主站和从站角色的探针:扮演主站时主动去读设备里的线圈和寄存器,扮演从站时模拟一个真实设备,让正在开发的主站程序或PLC有东西可试。它的价值集中在三件事——验证连通性、摸清寄存器映射、压出稳定性和超时问题。适合做自动化集成、PLC联调和设备出厂验证的人直接照搬。
2. modbusTCP报文底层:先把一帧请求和响应的结构吃透
如果只把测试软件当按钮工具用,遇到半包、异常码、事务ID错乱就会直接翻车。协议层面不搞清楚,后面所有参数调整都是在瞎猜。所以第一件事不是写代码,而是把一个Modbus TCP请求、一个响应,按字节拆开看明白。
2.1 为什么测试要分清主站和从站:502端口与连接方向决定一切
Modbus TCP本质上是把串口Modbus封装进TCP,默认端口502。主站(Master)是发起连接的一侧,也叫客户端;从站(Slave)是被动监听的一侧,也就是服务端。测试软件在切换这两种角色时,连接方向是反的:主站模式下软件主动去连设备,从站模式下软件要自己绑定502端口等别人的连接来。
这个区别直接决定了测试软件的资源管理策略。主站模式下,软件要为每一个目标设备维护一条TCP连接,还要处理超时、重连、半包。从站模式下,软件要监听端口、处理并发连接,同时保证在收到异常报文时能回一个规范的错误响应。很多初写的测试工具只把报文拼对了,却不管理连接,结果就是轮询几百次后连接假死,怎么调都调不好。
我一般会建议测试工具把这两种角色拆成两个独立入口,不要放在同一个线程里跑。原因很简单:主站模式要控制轮询节奏,从站模式要被动等待,两者对阻塞和超时的处理方式完全不同。混在一起写,往往会出现“从站一启动,主站就卡住”的尴尬局面。
2.2 功能码和数据地址映射:点到点测试的最小指令集怎么选
Modbus TCP的功能码决定了这条报文是读还是写、读写什么对象。测试软件不需要实现所有功能码,但必须清楚哪些是现场最常用的。这里列一张对照表,抄参数的时候会频繁用到:
| 功能码 | 名称 | 常见对象 | 手册寄存器区 |
|---|---|---|---|
| 0x01 | 读线圈 | 开关量输出 | 00001-09999 |
| 0x02 | 读离散输入 | 开关量输入 | 10001-19999 |
| 0x03 | 读保持寄存器 | 可读可写参数 | 40001-49999 |
| 0x04 | 读输入寄存器 | 只读测量值 | 30001-39999 |
| 0x05 | 写单个线圈 | 启停、开关 | 00001-09999 |
| 0x06 | 写单寄存器 | 设定单个参数 | 40001-49999 |
| 0x0F | 写多个线圈 | 批量开关输出 | 00001-09999 |
| 0x10 | 写多个寄存器 | 批量下发参数 | 40001-49999 |
测试软件的最小指令集,我建议只做五个功能码:0x01、0x03、0x05、0x06、0x10。原因是线圈类设备只涉及前三个,寄存器类设备覆盖后三个,中间的0x02和0x04大多数设备用不到。先把这几个调通,再根据需要补齐其他功能码,不要一开始就追求全功能。
地址映射是另一个高频坑。协议字段里的数据地址是从0开始的,但设备手册里经常写40001、40002这种1起始的地址。比如手册说“保持寄存器40001”,在报文里的实际地址就是0x0000;手册说“40011”,报文里要填0x000A。测试软件如果直接把手册地址填进去,读出来的永远是错位的值。
2.3 拆一个实际的读写请求:从抓包hex看MBAP帧的每个字节
抓包是排查Modbus TCP问题最直接的路径。下面是一帧“读10个保持寄存器,从0x0000开始”的完整请求报文:
00 01 00 00 00 06 01 03 00 00 00 0A逐字段拆开来看:前面6个字节是MBAP头,00 01是事务ID,00 00是协议ID(Modbus固定为0),00 06是长度字段,表示从单元ID开始到报文结尾一共还有6个字节。后面依次是01单元ID、03功能码、00 00起始地址、00 0A寄存器数量。整条报文控制在12字节,非常紧凑。
对应的响应如果读取成功,会是这种形式:
00 01 00 00 00 17 01 03 14 ...这里的00 17等于十进制的23,计算方式是单元ID占1字节、功能码占1字节、字节数占1字节、数据占20字节,加起来刚好23。也就是说,长度字段是从单元ID开始算的,不是从报文开头算的。
解析响应时有个关键点:TCP是流协议,不是串口帧,recv一次不一定能收到完整报文。下面这个函数按字节数切包,同时处理异常响应:
def parse_response(frame: bytes): if len(frame) < 9: return {"error": True, "reason": "帧长不足"} txn = int.from_bytes(frame[0:2], "big") proto = int.from_bytes(frame[2:4], "big") length = int.from_bytes(frame[4:6], "big") unit_id = frame[6] func = frame[7] # 功能码最高位置1表示异常响应 if func & 0x80: return {"error": True, "txn": txn, "unit_id": unit_id, "exception": frame[8]} byte_count = frame[8] registers = [ int.from_bytes(frame[9 + i*2: 11 + i*2], "big") for i in range(byte_count // 2) ] return {"txn": txn, "proto": proto, "length": length, "unit_id": unit_id, "func": func, "registers": registers}函数的逻辑是先确认帧长,再取事务ID和长度字段,然后判断功能码最高位。如果最高位是1,说明从站返回的是异常码而不是正常数据,这时第9字节就是异常码,不需要再解析寄存器。正常响应才读取字节数和寄存器列表,按每2字节一个大端整数切分。
参数说明:frame是完整的一帧响应,必须在调用这个函数之前按长度字段攒够字节;txn用于和请求对号入座,并发场景下不能用固定值;byte_count表示后面跟随的数据字节数,读完这么长就算完整解析了一帧。测试软件里最常见的半包问题,就是直接recv一次以为收完整了,结果下一条数据还在路上。
3. 从零写一个能用的modbusTCP测试工具:主站轮询与从站模拟的最小代码
原理讲完,接下来就是动手。这里的代码不是要替代一个完整的商业测试软件,而是给你一个能跑、能改、能排查的最小框架。在这个基础上加界面、加日志、加数据解析,就是一套自己的测试工具。
3.1 选型:自己用socket写还是套现成modbus库
写Modbus TCP测试工具,业界有种现成的偷懒办法:直接用某个成熟的Python Modbus通信库,几行代码就能连从站、读寄存器。库确实快,但对测试软件来说有个致命问题——它像个黑匣子。遇到异常响应、半包、连接假死,你没法分辨是库的问题还是设备的问题。我有过血泪经验:一次现场测设备,搞了整晚都以为是设备不回包,最后发现是库在TCP连接超时后没有主动重连,换回原生socket一测,设备根本没问题。
所以我一般建议先用标准socket库写最小一版,把MBAP头拼装、长度切包、异常码判断这几件事亲自做一遍。这套代码量不大,但能让你在出问题时从容排查。等验证完协议行为,再决定要不要切换到现成库来做批量轮询。对于测试软件这个场景,原生socket的灵活度比开发效率更重要。
3.2 主站模式:单次读写与循环轮询的核心代码
主站模式的核心函数有两类:一类是收发一个完整请求,另一类是循环轮询一批地址。先写一个严格按长度收包的辅助函数,这是避免半包问题的关键:
import socket import struct import time def recv_exact(sock, n, timeout): sock.settimeout(timeout) buf = b"" while len(buf) < n: try: chunk = sock.recv(n - len(buf)) except socket.timeout: raise TimeoutError("接收超时,连接可能已被对端关闭") if not chunk: raise ConnectionError("连接被对端关闭") buf += chunk return bufrecv_exact的要点是循环收,直到收够n字节。sock.recv只保证返回一部分字节,尤其是在跨网段的网络环境里,数据经常分片到达。如果不看长度硬收,后续报文就会被残留数据污染。
有了这个基础,写读保持寄存器的主站函数:
def read_holding_registers(ip, port, unit_id, start_addr, count, timeout=2.0, retries=3): last_err = None for attempt in range(retries): sock = None try: sock = socket.create_connection((ip, port), timeout=timeout) txn = 1 + attempt # 事务ID随重试递增,方便对日志 length = 6 # unit_id 1字节 + 功能码1字节 + 地址2字节 + 数量2字节 mbap = struct.pack(">HHHB", txn, 0, length, unit_id) pdu = struct.pack(">BHH", 0x03, start_addr, count) sock.sendall(mbap + pdu) header = recv_exact(sock, 6, timeout) txn_resp, proto_resp, body_len = struct.unpack(">HHH", header) body = recv_exact(sock, body_len, timeout) func = body[1] if func & 0x80: raise ValueError(f"从站异常,异常码: {body[2]}") byte_count = body[2] regs = [ int.from_bytes(body[3 + i*2: 5 + i*2], "big") for i in range(byte_count // 2) ] return regs except Exception as e: last_err = e if sock is not None: sock.close() time.sleep(1.0) # 重试间隔 print(f"读取失败: {ip}:{port} unit={unit_id} 原因={last_err}") return None注意这里的几处关键参数。timeout同时作用于connect和后续recv,默认2秒对于本地现场网络够用;如果设备响应在大多情况下是几十毫秒,2秒不至于误报。retries默认3次,每次重试都会新建TCP连接,而不是重用可能已经假死的连接。事务ID每次重试加1,在日志里能清楚看到哪次请求对应哪次响应。
循环轮询的代码可以很轻量:
def poll_devices(targets): for ip, unit_id in targets: for addr in range(0, 100, 10): regs = read_holding_registers(ip, 502, unit_id, addr, 10) if regs is None: print(f"[{ip}] unit={unit_id} 起始地址 {addr} 超时") else: print(f"[{ip}] unit={unit_id} 起始地址 {addr} 读到 {len(regs)} 个寄存器: {regs}") time.sleep(0.2)轮询间隔time.sleep(0.2)和超时是两个独立概念。超时是在等待响应,轮询间隔是两次请求之间的主动休眠。如果把轮询间隔设得比设备响应周期还短,除了压满网络没有意义;如果设得和超时一样长,设备一旦慢一点,整个测试会陷入超时、重试、再超时的死循环。
3.3 从站模式:模拟设备响应与单元ID映射的坑
从站模式的代码比主站更绕,因为它要做的是“收到什么就回什么”。下面这个最小从站绑定502端口,只处理读保持寄存器0x03,其他功能码一律回异常:
def start_slave(port=502): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("0.0.0.0", port)) server.listen(5) print(f"从站已启动,监听端口 {port}") while True: conn, _ = server.accept() try: header = recv_exact(conn, 7, 3.0) # 6字节MBAP + 1字节单元ID txn, proto, length, unit_id = struct.unpack(">HHHB", header) body = recv_exact(conn, length - 1, 3.0) # 长度已包含unit_id func = body[0] if func == 0x03: start_addr, qty = struct.unpack(">HH", body[1:5]) if start_addr + qty > 100: resp = build_response(txn, unit_id, 0x83, b"\x02") else: reg_data = b"".join( ((0x1234 + start_addr + i) & 0xFFFF).to_bytes(2, "big") for i in range(qty) ) resp = build_response(txn, unit_id, 0x03, reg_data) conn.sendall(resp) else: resp = build_response(txn, unit_id, func | 0x80, b"\x01") conn.sendall(resp) except Exception as e: print(f"连接处理异常: {e}") finally: conn.close()这里有一个容易搞混的点:first读取7字节是因为MBAP头本身6字节,后面紧跟着单元ID 1字节。而length字段是从单元ID开始计数的,所以读完7字节后,还需要再读length减去1的剩余字节。这个细节如果写错,最常见的表现是能建立连接但总是解析不到完整请求。
build_response这个函数把响应拼装逻辑收拢在一起,保证所有分支都走同样的MBAP构造规则:
def build_response(txn, unit_id, func, data): # 长度字段从unit_id开始计数 mbap = struct.pack(">HHHB", txn, 0, len(data) + 2, unit_id) return mbap + bytes([func]) + data参数说明:txn必须原样复制请求里的事务ID,否则主站会认为这条响应不是给自己的;unit_id也要原样回写,即使请求里的单元ID和从站自身不一致。很多真实设备在收到不匹配的单元ID时,会返回异常码02或者干脆不回,测试软件做成从站时最好也保留这个行为,否则就模拟不出现场的坑。
4. 必调参数:超时、重试、字节序与寄存器地址的边界
纸上代码跑通之后,真正耗时间的是调参。Modbus TCP测试软件翻车,十有八九不是协议没实现,而是下面几个参数没设对。
4.1 超时和重试策略:稳定性与误报之间的平衡
timeout设小了误报“设备无响应”,设大了整个回归测试被拖慢。给一套我常用的参考值:
| 场景 | 超时时间 | 重试次数 | 重试间隔 |
|---|---|---|---|
| 同一交换机下的设备 | 1-2秒 | 3次 | 0.5-1秒 |
| 跨网关或无线链路 | 3-5秒 | 3次 | 1秒 |
| 批量扫描大量从站 | 1秒 | 1次 | 0.2秒 |
这里的原则是:超时要能覆盖最慢一次正常响应,而不是覆盖最快一次。设备响应通常是20到50毫秒,但中间经过网关、无线、PLC调度后,偶发一两个500毫秒的响应很正常。给1秒以上不会误伤,给500毫秒反而会频繁触发重试。
重试间隔也很关键。连续重试之间至少要留一个喘息时间,否则设备本来能恢复,却被密集请求堵死。所有重试都应该新建连接,而不是复用失败的那条TCP链路。我见过不少测试软件在连接超时后,仍然往旧socket上发数据,结果每次都是立即报错,日志白茫茫一片。
4.2 字节序与32位数据组包:INT32和FLOAT翻车重灾区
Modbus协议规定单个16位寄存器内部是大端存储,但32位数据是两个寄存器拼出来的,厂商对拼接顺序的约定各不相同。常见的是高字在前,也有相当一部分设备是低字在前。测试软件如果不提供这个切换开关,读FLOAT和INT32几乎必错。
下面这段代码把两个寄存器拼成FLOAT或INT32,word_order参数控制字序:
import struct def regs_to_float(hw, lw, word_order="big"): if word_order == "big": raw32 = (hw << 16) | lw else: raw32 = (lw << 16) | hw return struct.unpack(">f", raw32.to_bytes(4, "big"))[0] def regs_to_int32(hw, lw, word_order="big"): if word_order == "big": raw32 = (hw << 16) | lw else: raw32 = (lw << 16) | hw return raw32 - 0x100000000 if raw32 & 0x80000000 else raw32参数说明:hw是高16位对应的寄存器值,lw是低16位对应的寄存器值。word_order传入big时按高字在前组装,传入little时按低字在前组装。调到这个函数里的值,应该已经从报文字节里按大端读出来了,不要再做二次反转。
验证字节序有个很好的参照:IEEE754里浮点数1.0的32位表示是0x3F800000。也就是说,正常大端字序下,第一个寄存器是0x3F80,第二个是0x0000。如果设备手册说这是FLOAT,但两个寄存器反过来,那就要用little字序。这个已知值能帮你快速判断设备到底走的哪种规则,比对着设备上位机数据猜快得多。没有后悔药,只能把所有寄存器原始值打印成HEX,再和上位机软件对照。
4.3 寄存器地址0/1基准:批量读写数量上限不能碰
前面提过,协议地址从0开始,手册地址常从1开始。这种0和1的差在单个寄存器上看着不起眼,一进入批量读写就全乱。我习惯在测试软件里加一个内部换算函数,把手册地址统一转换成协议地址:
def doc_addr_to_wire(doc_addr, base=40001): return doc_addr - base参数说明:doc_addr是设备手册里的寄存器编号,比如40011;base是手册那个区间的起始编号,保持寄存器通常从40001开始。这个函数只做减法,不做其他处理,保证所有报文统一填协议地址,界面统一显示手册地址。
批量读写还要注意协议的上限。读保持寄存器0x03一次最多125个寄存器,写多个寄存器0x10一次最多123个。这是Modbus协议PDU长度限制决定的,超过这个数量,设备会返回异常码03或02。测试软件如果做大批量扫描,要自己分块:
MAX_READ_REGS = 125 MAX_WRITE_REGS = 123 def split_ranges(start, count, max_count, max_value=65535): if start + count > max_value: count = max_value - start for off in range(0, count, max_count): yield start + off, min(max_count, count - off)split_ranges一次给出一段不超过max_count的小区间,循环调用就能把一个大范围拆成多次请求。别小看这个函数,很多现场“一读就异常,单读却正常”的玄学问题,就是因为工具在挂扫描时没有做分块,一条请求超过125个寄存器,设备直接回了非法地址。
5. modbusTCP测试过程中的常见问题与排查:五条真实踩坑记录
调参只能解决一半问题,另一半藏在现场的各种怪象里。这里记录五条我实际踩过的坑,每一条都按现象、原因、解决三步写清楚。
5.1 现象:从站连上但读取返回异常码02
现象:TCP连接建立成功,主站发送读请求,从站立刻回应一个异常响应,功能码变成0x83,异常码是0x02。
原因:0x02表示非法数据地址。最常见的是起始地址超出了从站的寄存器映射范围,或者请求的寄存器数量跨过了合法区间的末尾。另一种情况是,测试工具把手册里40001这个编号直接当作协议地址0填了进去,导致整个访问错位一个地址。
解决:先把读取范围缩小到单个地址,从0开始试。确认这一条能读通之后,再逐步扩大范围,直到复现异常码。同时检查设备手册里的寄存器总长度,确认地址映射是0起始还是1起始。如果是1起始,用4.3里那个换算函数统一处理。
5.2 现象:轮询一段时间后请求超时,重连又恢复
现象:测试软件刚启动时一切正常,轮询几百次后突然连续超时,停掉再重新运行又好了,或者代码里加了断线重连之后就恢复。
原因:TCP连接被中间网络设备回收了。现场交换机、网关、防火墙对空闲连接普遍有超时机制,几十秒到几分钟不等。测试软件的轮询间隔如果较长,连接可能处于“看起来还在,实际已被拆掉”的状态。
解决:在应用层做保活,每30秒至少主动发送一次请求,而不是靠系统自带的keepalive。系统级TCP keepalive在很多平台上参数不可调,Windows下尤其别扭。更稳妥的做法是,每次请求失败后直接关闭旧连接并新建,不要在旧socket上反复重发。这也是上一章主站代码里每次重试都新建连接的原因。
5.3 现象:本机502端口被占用,从站一直起不来
现象:从站模式启动时报端口绑定失败,换端口又能起来,但换回502就报错。
原因:502是Modbus TCP默认端口,也是很多工具软件默认监听的位置。同机可能运行了多个测试实例,或者有别的服务占用了502。另外在Linux较老版本上,1024以下端口需要root权限;如果从站跑在容器里,端口映射和主机端口冲突也会导致这个现象。
解决:先查占用情况,Windows用netstat -ano | findstr :502,Linux用lsof -i :502,找到占用进程后停掉或改端口。如果只是测试环境,直接让从站监听1502这种高位端口,主站侧把目标端口改成对应值就行。协议本身端口可配置,测试软件里不要写死502。
5.4 现象:读到数值与设备上位机不一致,正负还反了
现象:同一台设备,设备自带的上位机软件显示温度25.0,而测试软件自己读出来的原始寄存器值转成小数后是-0.0或者完全不对的数。
原因:大概率是数据类型和字节序不匹配。寄存器0xFFFF既可能是无符号65535,也可能是带符号-1,取决于设备手册按INT16还是UINT16解释。如果是INT32或FLOAT,还要看两个寄存器的字序,不在意的话解析出来完全不可用。
解决:先读原始寄存器值,打印成HEX格式,和上位机软件的值做对照。用4.2节那个已知值判断字节序,再在测试软件里加上数据类型选择:UINT16、INT16、UINT32、INT32、FLOAT32,每种类型对应不同的解析函数。符号问题多半是把INT16当UINT16读了。
5.5 现象:偶发超时,抓包却有响应,代码却报错
现象:测试软件里偶发报错,抓包却能看到从站真实返回了正常响应,时间也很及时,但软件还是把这次请求判成失败。
原因:TCP流的问题,不是设备问题。响应可能被拆成两个TCP段,第一次recv只收到一半,测试软件却当完整帧去解析,解析失败就报超时。还有一种情况是并发场景下多个请求共用一个连接,响应的顺序可能和请求顺序不一致,程序拿本次请求的事务ID去对不上号,就出现了“有响应但软件说超时”。
解决:所有收包都用recv_exact按长度切包,没有哪段是可以只recv一次就完事的。并发测试时,每个请求使用独立的事务ID,并且按事务ID去匹配响应,不能假设先发先回。如果只是简单轮询,干脆每轮请求都用独立连接,彻底避开乱序问题。
6. 进阶:把modbusTCP测试脚本变成能过夜的自动化回归工具
手工点按钮测试只能验证“现在通不通”,做不了“能不能稳定跑过夜”。把这个最小工具往自动化方向走,只需要加日志和统计两件事。
6.1 给每个请求打时间戳:用日志定位主站、网络还是设备抖动
在轮询循环里记录请求开始时间和耗时,而不是只在失败时打印一行错误。下面是插入到主站模式里的记录逻辑:
start = time.perf_counter() regs = read_holding_registers(ip, 502, unit_id, addr, 10) cost_ms = (time.perf_counter() - start) * 1000 records.append({ "ts": time.strftime("%Y-%m-%d %H:%M:%S"), "ip": ip, "unit": unit_id, "addr": addr, "cost_ms": round(cost_ms, 2), "ok": regs is not None })这样每条记录都带时间戳和耗时,偶发超时就不再是一团乱麻。如果慢请求的时间点和网络重传时间点重合,问题在网络;如果慢请求集中在某个设备上,问题在设备负载或链路质量;如果所有设备在同一时刻变慢,要去看交换机和网关。
6.2 导出CSV与阈值比较:让测试结论不靠人肉盯屏
把records逐行写入CSV,然后按设备分组统计成功率和响应时间。我最后的收尾方式是写了一个阈值判断:成功率必须大于等于99.9%,P95响应时间小于200毫秒,否则整个测试判失败。
import csv with open("modbus_regression.csv", "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=["ts", "ip", "unit", "addr", "cost_ms", "ok"]) writer.writeheader() writer.writerows(records)自动跑完后只看这个CSV,不再需要人盯着屏幕看一晚上。我现在做现场测试的习惯是:连接失败先重连,并把这次失败记录成一条完整日志,而不是直接退出程序;整晚跑完后,按设备、按地址段汇总统计,再和前一晚的数据对比。这个习惯让我少背了很多电量锅,也让客户能拿着同一份报告去和供货商对线。偶发问题最玄学,但把时间戳、耗时、原始报文都记录下来之后,就不玄了。希望帮到你。
本文还有配套的精品资源,点击获取