简介:这份资源是面向工业自动化与上位机开发者的汇川PLC与C# Modbus TCP通信测试工程,适合已具备一定C#基础、希望打通PLC数据采集与控制链路的工程师学习参考。压缩包共47个文件,约922KB,以cs源码、resx与resources界面资源、config配置、exe可执行程序及sln/csproj工程文件为主,另含少量pdb调试符号与缓存文件,可直接用Visual Studio打开运行。内容围绕Modbus TCP协议封装、Socket非阻塞连接、功能码读写、异常重连与数据解析显示等关键环节展开,配套AM402控制器测试程序,便于对照理解请求构造与响应解析的完整流程。目前已有2254人学习下载,可作为搭建Winform上位机、排查通信超时与断线重连问题的实用参考。
1. Modbus TCP 通讯测试:从压缩包名到可复现的调试链路
拿到一个叫InproModebusTCP通讯测试.rar的包,第一反应不该是解压看代码,而是先想清楚它要解决什么问题。Modbus TCP 是工业现场最常见的以太网通讯协议之一,把传统的 Modbus RTU 串口报文塞进 TCP/IP 协议栈里传输,端口号默认 502。所谓“通讯测试”,本质就是验证客户端和服务端能不能按 Modbus 协议把寄存器读写跑通——读线圈、读保持寄存器、写单个寄存器这些功能码能不能正常收发。这个方向适合做 PLC 数据采集、工控网关、边缘计算盒子对接设备的工程师,也适合手里有 Inpro 相关设备或软件、需要验证链路是否通畅的现场调试人员。下面按“协议先立住、再动手复现、最后排坑”的顺序讲透。
2. Modbus TCP 协议栈拆解:报文结构、功能码与端口约定
2.1 Modbus TCP 和 Modbus RTU 的报文差异
Modbus RTU 的帧结构是「从站地址 + 功能码 + 数据 + CRC 校验」,靠串口的静默间隔来分帧。Modbus TCP 去掉了从站地址和 CRC,换成了 7 字节的 MBAP 头(Modbus Application Protocol Header),结构如下:
| 字段 | 长度 | 说明 |
|---|---|---|
| 事务标识符 | 2 字节 | 客户端递增,用于匹配请求和响应 |
| 协议标识符 | 2 字节 | Modbus 固定为 0x0000 |
| 长度 | 2 字节 | 后续字节数(单元标识符 + PDU) |
| 单元标识符 | 1 字节 | 类似从站地址,网关场景常用 |
| 功能码 | 1 字节 | 如 0x03 读保持寄存器 |
| 数据 | N 字节 | 起始地址、寄存器数量等 |
关键点:TCP 是流式协议,没有消息边界。Modbus TCP 靠 MBAP 头里的「长度」字段来界定一帧的结束,这就是它解决粘包问题的天然方案。很多人从串口转过来会不习惯——RTU 靠时间间隔分帧,TCP 靠长度字段分帧,这是两套完全不同的思路。
2.2 常用功能码与地址映射
现场调试 90% 的场景只用四个功能码:
- 0x01 读线圈:读开关量输出,地址范围 00001-09999
- 0x02 读离散输入:读开关量输入,地址范围 10001-19999
- 0x03 读保持寄存器:读模拟量参数,地址范围 40001-49999
- 0x04 读输入寄存器:读只读模拟量,地址范围 30001-39999
这里有个血泪经验:协议文档里的「40001」是给人看的 1-based 地址,实际报文里填的是 0-based 偏移量。比如要读 40001,报文里起始地址填 0x0000;读 40010,填 0x0009。搞错这个偏移,读回来的数据永远对不上,而且不会报错,只是数据错位,非常隐蔽。
2.3 端口号与连接模型
Modbus TCP 默认端口 502,这是 IANA 分配的标准端口。服务端监听 502,客户端发起连接。一个 TCP 连接上可以连续发多帧 Modbus 请求,靠事务标识符区分。常见做法是客户端保持长连接,定时轮询;也有短连接模式,每次请求建连再断开,适合请求频率极低的场景。长连接效率高但需要处理断线重连,短连接简单但开销大。我一般用长连接加心跳,下面会讲怎么实现。
3. 用 Python 搭一套 Modbus TCP 测试环境
3.1 服务端:用 pymodbus 模拟从站
先装库:
pip install pymodbus写一个最小服务端,模拟一个带保持寄存器和线圈的从站:
from pymodbus.server import StartTcpServer from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext # 初始化数据块:线圈100个,保持寄存器100个 # 参数说明:0x00 是起始地址,[0]*100 是初始值列表 store = ModbusSlaveContext( co=ModbusSequentialDataBlock(0x00, [0] * 100), # 线圈 hr=ModbusSequentialDataBlock(0x00, [100] * 100), # 保持寄存器,初值100 ) context = ModbusServerContext(slaves=store, single=True) # 监听所有网卡的502端口 # 参数说明:address 是绑定地址,port 是端口,context 是数据上下文 StartTcpServer(context=context, address=("0.0.0.0", 502))逻辑说明:ModbusSequentialDataBlock创建连续地址空间,第一个参数是起始地址,第二个是初始值列表。single=True表示只有一个从站,单元标识符不区分。服务端启动后会在 502 端口监听,任何符合 Modbus TCP 格式的请求都会得到响应。
参数说明:保持寄存器初值设成 100 是为了方便验证——客户端读回来应该全是 100。线圈初值全 0,写一个 1 再读回来能验证写功能。
3.2 客户端:读保持寄存器和写线圈
from pymodbus.client import ModbusTcpClient # 连接服务端 # 参数说明:host 是服务端IP,port 是端口,timeout 是超时秒数 client = ModbusTcpClient("127.0.0.1", port=502, timeout=3) client.connect() # 读保持寄存器:从地址0开始读10个 # 参数说明:address 是起始地址(0-based),count 是寄存器数量,slave 是从站ID rr = client.read_holding_registers(address=0, count=10, slave=1) print("保持寄存器:", rr.registers) # 预期 [100]*10 # 写单个线圈:地址5写True wr = client.write_coil(address=5, value=True, slave=1) print("写线圈结果:", wr) # 读回线圈验证 rc = client.read_coils(address=5, count=1, slave=1) print("线圈状态:", rc.bits[0]) # 预期 True client.close()逻辑说明:read_holding_registers发的是功能码 0x03,write_coil发的是 0x05。注意address参数是 0-based,对应协议里的偏移量。slave参数对应 MBAP 头里的单元标识符,单从站场景填 1 即可。
参数说明:timeout=3是连接和读写超时,现场网络差可以调到 5。count一次最多读 125 个寄存器(协议限制),超了会报异常。
3.3 用 tcpdump 或 Wireshark 抓包验证
光看代码返回不够,要确认线上跑的报文对不对。Linux 下用 tcpdump:
# 抓502端口,写文件供Wireshark分析 # 参数说明:-i any 监听所有网卡,-w 写文件,port 502 过滤端口 sudo tcpdump -i any -w modbus.pcap port 502Windows 下直接开 Wireshark,过滤条件填tcp.port == 502。抓到的包展开 MBAP 头,重点看三个地方:事务标识符是否递增、长度字段是否等于后续字节数、功能码和数据是否符合预期。如果长度字段对不上,说明发送端组包有问题;如果事务标识符不匹配,说明响应和请求对不上号,可能是并发请求没处理好。
4. 现场调试避坑:连接、粘包与超时的排查记录
4.1 连接被拒绝或超时
现象:客户端connect()返回 False,或者读写超时。
原因:三种可能——服务端没启动、防火墙拦了 502 端口、IP 或端口填错。工业现场还多一种:设备只允许特定源 IP 访问,或者端口被改成非 502。
解决:先在服务端机器上netstat -an | grep 502确认监听状态。再从客户端telnet 服务端IP 502测连通性。Windows 防火墙用netsh advfirewall firewall add rule name="Modbus" dir=in action=allow protocol=TCP localport=502放行。如果设备改过端口,用 nmap 扫一下常见端口段。
4.2 粘包导致数据错位
现象:连续发多帧请求,响应数据串了,或者解析出乱码。
原因:TCP 是流式协议,如果自己手写 socket 收发,没有按 MBAP 头的长度字段分帧,就会把两帧粘在一起。常见于用 C 语言裸写 socket 的场景。
解决:接收缓冲区里先读 7 字节 MBAP 头,解析出长度字段 N,再读 N 字节。循环这个过程。用 pymodbus 这类库不用操心,库内部已经处理了。自己写的话,参考下面逻辑:
def recv_modbus_frame(sock): # 先读7字节MBAP头 header = b"" while len(header) < 7: chunk = sock.recv(7 - len(header)) if not chunk: return None header += chunk # 解析长度字段(第5-6字节) length = int.from_bytes(header[4:6], "big") # 再读length字节的PDU body = b"" while len(body) < length: chunk = sock.recv(length - len(body)) if not chunk: return None body += chunk return header + body参数说明:header[4:6]是 MBAP 头的长度字段,大端序。length包含单元标识符和 PDU,所以后续要读这么多字节。
4.3 事务标识符不匹配
现象:并发请求时,响应和请求对不上,读回来的数据是别的请求的。
原因:多个请求共用一个连接,事务标识符没有正确递增或匹配。有些简易客户端固定事务标识符为 0,并发场景就乱了。
解决:每个请求分配唯一的事务标识符,收到响应后按标识符匹配。pymodbus 内部用锁保证串行,不会出现这个问题。自己实现的话,维护一个递增计数器,响应回来先查表再处理。
4.4 寄存器地址偏移搞错
现象:读 40001 读回来的是 40002 的值,或者全零。
原因:协议文档用 1-based 地址,报文用 0-based 偏移。40001 对应偏移 0,40002 对应偏移 1。搞混了就整体错一位。
解决:记住公式「报文地址 = 文档地址 - 40001」。写代码时在注释里标清楚,别靠脑子记。测试时先读一个已知值的寄存器验证偏移对不对。
4.5 长连接断线没重连
现象:跑了一晚上,第二天发现数据不更新了,但程序没报错。
原因:网络抖动或设备重启导致 TCP 连接断开,客户端没有检测机制,还在往一个死连接上写数据。
解决:加心跳和重连。每次读写前检查client.connected,断了就重连。或者定时发一个读请求当心跳,失败就触发重连。下面是一个简单的重连封装:
import time from pymodbus.client import ModbusTcpClient class ModbusWrapper: def __init__(self, host, port=502): self.host = host self.port = port self.client = None self.connect() def connect(self): # 参数说明:retries=3 重试3次,timeout=3 超时3秒 self.client = ModbusTcpClient(self.host, port=self.port, timeout=3) self.client.connect() def read_hr(self, address, count, slave=1): try: if not self.client.connected: self.connect() rr = self.client.read_holding_registers(address, count, slave=slave) if rr.isError(): raise Exception("Modbus error") return rr.registers except Exception as e: print(f"读取失败: {e},重连中...") time.sleep(1) self.connect() return None逻辑说明:每次读之前检查连接状态,断了就重连。读失败也触发重连。isError()判断 Modbus 异常响应(比如非法地址)。参数说明:timeout=3根据现场网络质量调整,无线网络可以放到 5。
5. 进阶技巧:用脚本做批量回归与性能压测
单次读写跑通只是第一步,真正上线前要做批量回归和性能压测。我一般写一个脚本,把所有要读的寄存器地址列成表,循环读一遍,记录每个地址的响应时间和值,输出成 CSV。这样既能验证地址映射对不对,又能看出哪些地址响应慢。
import csv import time from pymodbus.client import ModbusTcpClient # 待测地址表:起始地址、数量、名称 test_points = [ (0, 10, "温度寄存器组"), (20, 5, "压力寄存器组"), (50, 20, "状态寄存器组"), ] client = ModbusTcpClient("192.168.1.100", port=502, timeout=3) client.connect() with open("modbus_test_result.csv", "w", newline="") as f: writer = csv.writer(f) writer.writerow(["名称", "起始地址", "数量", "耗时ms", "值"]) for addr, count, name in test_points: t0 = time.time() rr = client.read_holding_registers(addr, count, slave=1) elapsed = (time.time() - t0) * 1000 if rr.isError(): writer.writerow([name, addr, count, f"{elapsed:.1f}", "ERROR"]) else: writer.writerow([name, addr, count, f"{elapsed:.1f}", rr.registers]) time.sleep(0.05) # 间隔50ms,避免打爆设备 client.close()逻辑说明:time.time()记录每次读的耗时,isError()捕获异常响应。time.sleep(0.05)是请求间隔,工业设备 CPU 弱,连续高频请求可能丢包,加个间隔更稳。参数说明:间隔根据设备性能调,PLC 一般 20-50ms 没问题,单片机网关可能要 100ms 以上。
压测的话,把间隔去掉,循环发 1000 次,统计平均耗时和丢包率。丢包率超过 1% 就要查网络或设备负载。我踩过的坑是:压测时用短连接,每次建连开销大,测出来耗时偏高,误判成设备慢。后来改成长连接,数据才准。
还有一个技巧:用 Wireshark 的tcp.time_delta字段看请求和响应之间的时间差,这个才是设备真实的处理时间,比在应用层测的准。应用层测的包含了 Python 解释器开销和网络栈开销。
最后说个习惯:每次现场调试,先抓包再写代码。抓包能看到最原始的报文,确认设备到底支不支持某个功能码、地址偏移是多少、响应格式对不对。靠猜和试,效率太低。希望帮到你。
本文还有配套的精品资源,点击获取