Modbus TCP服务端模拟器：从协议解析到并发调试的工程实践

简介：ModbusTcpServer1.zip 是一款面向工业自动化与工业物联网开发者的 Modbus TCP 服务端模拟器，适合需要在无真实 PLC 硬件条件下调试通信程序的 C# 工程师与学习者。它可模拟输入寄存器、保持寄存器、离散输入和线圈等设备行为，帮助在开发阶段验证客户端读写逻辑，降低联调成本。压缩包共 9 个文件，约 928KB，以 exe 可执行程序为主，辅以 dll 通信库、pdb 调试符号与 xml 接口文档，便于直接运行并了解底层调用方式。目前已有 3276 人学习下载，说明其在工控通信入门与测试场景中具有一定参考价值。借助该模拟器，读者可搭建本地 Modbus TCP 服务端，配合 Socket 或 NModbus 等库编写客户端，完成连接建立、请求报文封装、响应解析与异常排查的完整练习，为实际工控项目中的设备监控与数据采集打下基础。

1. 一个压缩包名背后：ModbusTcpServer1.zip 到底能帮你解决什么

如果你在工控现场调过设备，大概率遇到过这种局面：PLC 还没到货，上位机软件已经写好了，组态画面也画完了，可就是没有从站可以连。你盯着空荡荡的设备列表，只能干等。ModbusTcpServer1.zip这个标题指向的，正是一个能让你在本地把 Modbus TCP 服务端跑起来的模拟器方案。它的价值不在于代码有多复杂，而在于让你在没有真实硬件的情况下，把通信链路先打通、把寄存器地址先对齐、把异常响应先验证一遍。适合谁用？写上位机采集程序的开发者、做 SCADA 组态调试的工程师、以及需要批量验证 Modbus 主站逻辑的测试人员。你不需要等设备到场，一台笔记本就能把从站角色扮演起来。

2. 先把 Modbus TCP 的报文结构吃透：从 MBAP 头到功能码

2.1 为什么 TCP 上还要套一层 MBAP

Modbus 原本是串行链路上的协议，跑到 TCP 上之后，最大的变化是多了 MBAP 头（Modbus Application Protocol Header）。很多人第一次抓包会懵：明明发的是读保持寄存器，怎么前面多了一串字节？这串字节就是 MBAP，共 7 个字节，作用是让接收方能从 TCP 流里切出一条完整的 Modbus 报文。

MBAP 的字段构成如下：

事务标识符这个设计很关键。主站可以连续发多条请求而不等待响应，靠事务标识符来区分哪条响应属于哪条请求。如果你写的服务端把事务标识符写死成 0，主站并发请求时就会乱套。常见做法是原样回传，不要自作主张改。

2.2 功能码不是随便定的：常用码与异常码

Modbus 的功能码决定了你要操作哪片寄存器区。做服务端模拟器，至少要支持以下几类：

• 0x01 读线圈：可读写布尔量，对应开关状态
• 0x02 读离散输入：只读布尔量，对应传感器通断
• 0x03 读保持寄存器：可读写 16 位字，最常用
• 0x04 读输入寄存器：只读 16 位字，对应模拟量采集
• 0x05 写单个线圈
• 0x06 写单个保持寄存器
• 0x0F 写多个线圈
• 0x10 写多个保持寄存器

当请求越界或功能码不支持时，服务端要返回异常响应。异常响应的功能码是原功能码加 0x80，后面跟一个异常码。比如请求读保持寄存器但起始地址超出范围，返回 0x83 加异常码 0x02（非法数据地址）。这个细节很多简易模拟器会忽略，直接不回或者回一个错误的功能码，导致主站侧超时而不是收到明确的异常，排查起来很费劲。

提示：异常码 0x01 表示非法功能，0x02 表示非法数据地址，0x03 表示非法数据值，0x04 表示从站设备故障。把这四个码用对，主站侧的错误处理逻辑才能被正确触发。

2.3 寄存器区的地址映射：0 基和 1 基的坑

Modbus 协议文档里经常出现 40001、30001 这种地址，这是 1 基的表示法，而且隐含了寄存器区。40001 表示保持寄存器的第 1 个寄存器，对应协议报文里的地址 0。很多新手在这里翻车：上位机填了 40001，服务端代码里却从地址 1 开始存，结果读出来的数据整体偏移一个寄存器。

我一般会在服务端内部统一用 0 基地址，然后在配置层做一个映射表，把用户看到的 40001 转成内部偏移 0。这样协议层干净，配置层灵活。如果你做的模拟器要给多人用，这个映射表最好做成可配置的，别写死在代码里。

3. 用 Python 把 ModbusTcpServer1.zip 的核心逻辑跑起来

3.1 最小可运行的服务端骨架

下面这段代码实现了一个最简 Modbus TCP 服务端，支持 0x03 和 0x06 两个功能码，监听 502 端口。你可以直接复制运行，然后用任何 Modbus 主站工具去连。

import socket import struct # 保持寄存器区，0基地址，初始化为100个寄存器 HOLDING_REGISTERS = [0] * 100 def handle_request(data): """解析MBAP头 + PDU，返回响应字节""" if len(data) < 8: return None # 解析MBAP头 trans_id, proto_id, length, unit_id = struct.unpack('>HHHB', data[:7]) if proto_id != 0: return None # 协议标识符必须为0 func_code = data[7] if func_code == 0x03: # 读保持寄存器：起始地址(2字节) + 寄存器数量(2字节) start_addr, reg_count = struct.unpack('>HH', data[8:12]) if start_addr + reg_count > len(HOLDING_REGISTERS): # 返回异常响应：功能码+0x80，异常码0x02 return build_response(trans_id, unit_id, bytes([0x83, 0x02])) values = HOLDING_REGISTERS[start_addr:start_addr + reg_count] payload = struct.pack('>' + 'H' * reg_count, *values) pdu = bytes([0x03, len(payload) * 2]) + payload return build_response(trans_id, unit_id, pdu) elif func_code == 0x06: # 写单个保持寄存器：地址(2字节) + 值(2字节) addr, value = struct.unpack('>HH', data[8:12]) if addr >= len(HOLDING_REGISTERS): return build_response(trans_id, unit_id, bytes([0x86, 0x02])) HOLDING_REGISTERS[addr] = value # 写单个寄存器的响应是原样回传请求的PDU return build_response(trans_id, unit_id, data[7:12]) else: # 不支持的功能码，返回非法功能异常 return build_response(trans_id, unit_id, bytes([func_code + 0x80, 0x01])) def build_response(trans_id, unit_id, pdu): """组装MBAP头 + PDU""" length = len(pdu) + 1 # +1是单元标识符 mbap = struct.pack('>HHHB', trans_id, 0, length, unit_id) return mbap + pdu def main(): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(('0.0.0.0', 502)) server.listen(5) print('Modbus TCP服务端已启动，监听502端口') while True: conn, addr = server.accept() print(f'客户端接入: {addr}') try: while True: data = conn.recv(1024) if not data: break resp = handle_request(data) if resp: conn.send(resp) except ConnectionResetError: pass finally: conn.close() if __name__ == '__main__': main()

这段代码的逻辑说明：handle_request负责解析 MBAP 头和功能码，根据功能码分发到不同处理分支。build_response负责组装响应报文，注意长度字段要包含单元标识符。main里用SO_REUSEADDR是为了避免重启时端口被占用报错。

参数说明：HOLDING_REGISTERS的长度决定了可访问的寄存器范围，你可以按需扩大。监听地址0.0.0.0表示接受任意网卡接入，如果只想本机测试可以改成127.0.0.1。502 是 Modbus TCP 的标准端口，非 root 用户在某些系统上可能无法绑定，可以临时改成 5020 测试。

3.2 用 pymodbus 快速验证你的服务端

写完服务端，怎么确认它是对的？我一般会用 pymodbus 写一个客户端脚本去读，比手动发字节快得多。

from pymodbus.client import ModbusTcpClient client = ModbusTcpClient('127.0.0.1', port=502) client.connect() # 写单个保持寄存器，地址0，值1234 client.write_register(0, 1234, unit=1) # 读保持寄存器，从地址0开始读4个 result = client.read_holding_registers(0, 4, unit=1) if result.isError(): print('读取失败:', result) else: print('寄存器值:', result.registers) client.close()

逻辑说明：先写后读，验证写入的值能否被正确读出。unit=1对应 MBAP 头里的单元标识符，你的服务端如果没校验这个字段，填什么都能通，但建议还是保持一致。

参数说明：write_register的地址是 0 基，对应协议报文里的地址。如果你习惯用 40001 这种表示法，记得在客户端侧做转换，别直接填 40001，否则会读到越界地址。

3.3 多客户端并发时要注意什么

上面的骨架是单线程的，一个客户端断开后才能接受下一个连接。实际测试中，你可能需要同时模拟多个从站或者让多个主站同时接入。最简单的改法是把handle_request放到线程里处理。

import threading def client_thread(conn, addr): try: while True: data = conn.recv(1024) if not data: break resp = handle_request(data) if resp: conn.send(resp) except ConnectionResetError: pass finally: conn.close() # 在main的accept循环里 while True: conn, addr = server.accept() t = threading.Thread(target=client_thread, args=(conn, addr), daemon=True) t.start()

逻辑说明：每个连接一个线程，互不阻塞。daemon=True让线程随主进程退出，避免程序卡住。

参数说明：线程模型下，HOLDING_REGISTERS是共享的，如果多个客户端同时写同一个寄存器，会有竞态。测试场景一般无所谓，但如果要模拟真实设备，建议加锁或者用队列串行化写操作。

4. 避坑指南：Modbus TCP 模拟器调试中最容易翻车的五个点

4.1 现象：主站报超时，但服务端日志显示已收到请求

原因：服务端处理完请求后没有发送响应，或者响应的 MBAP 头长度字段算错了，导致主站认为报文不完整一直等。

解决：在build_response里打印一下组装的字节，用 Wireshark 抓包对比。重点检查长度字段是否等于 PDU 长度加 1。我见过有人把长度算成 PDU 长度，少了单元标识符那一个字节，主站就一直等。

4.2 现象：读回来的数据整体偏移一个寄存器

原因：地址基准不统一。上位机用 1 基地址发请求，服务端按 0 基去取，或者反过来。

解决：在服务端入口处统一做一次转换，把所有请求地址减 1 或者加 1，确保内部存储只用一种基准。配置层给用户看的地址可以保留 40001 这种写法，但协议层必须干净。

4.3 现象：写单个寄存器成功，但写多个寄存器时主站报异常

原因：0x10 功能码的请求格式和 0x06 不同，多了字节计数和寄存器数量字段，解析时偏移量算错。

解决：0x10 的请求 PDU 是：功能码(1) + 起始地址(2) + 寄存器数量(2) + 字节计数(1) + 数据(N*2)。解析时先读起始地址和数量，再根据字节计数读数据。响应是：功能码(1) + 起始地址(2) + 寄存器数量(2)，共 5 字节。别把请求的格式套到响应上。

4.4 现象：服务端重启时报“地址已被占用”

原因：TCP 连接处于 TIME_WAIT 状态，端口还没释放。

解决：设置SO_REUSEADDR选项，就是上面代码里那行server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)。这个选项让操作系统允许绑定处于 TIME_WAIT 的地址，调试阶段必加。

4.5 现象：单元标识符不匹配导致主站丢弃响应

原因：有些主站会校验响应里的单元标识符是否和请求一致。如果你的服务端固定返回 0 或者 1，而主站发的是其他值，就会被丢弃。

解决：在build_response里把请求的单元标识符原样传回去，不要写死。这个字段在 TCP 场景下虽然常被忽略，但校验起来就是致命的。

5. 进阶：把模拟器做成可配置、可观测的调试工具

5.1 用 JSON 配置文件管理寄存器初始值

硬编码寄存器初始值只适合一次性测试。实际项目中，你可能需要模拟不同设备的寄存器映射。我一般会用一个 JSON 文件来描述：

{ "unit_id": 1, "holding_registers": { "0": 100, "1": 200, "10": 3000 }, "coils": { "0": true, "1": false } }

然后在启动时加载这个文件，把键值对填进对应的寄存器数组。这样换一个设备只需要换配置文件，不用改代码。注意 JSON 的键是字符串，读进来要转成整数再当索引。

5.2 加一个简单的 HTTP 接口查看当前寄存器状态

调试的时候，光看日志不够直观。我会在服务端里起一个极简的 HTTP 服务，返回当前所有寄存器的值。不需要框架，用 Python 标准库的http.server就够了。

from http.server import HTTPServer, BaseHTTPRequestHandler import json class StatusHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path == '/registers': self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(HOLDING_REGISTERS).encode()) else: self.send_response(404) self.end_headers() def start_http(): httpd = HTTPServer(('127.0.0.1', 8080), StatusHandler) httpd.serve_forever() # 在主程序里用线程启动 threading.Thread(target=start_http, daemon=True).start()

逻辑说明：访问http://127.0.0.1:8080/registers就能看到当前寄存器数组的 JSON 表示。调试时开着浏览器刷新，比翻日志快得多。

参数说明：HTTP 端口用 8080 避免和 Modbus 的 502 冲突。这个接口只绑定本机，不要暴露到公网，测试工具不需要考虑鉴权。

5.3 用单元测试锁定协议解析的正确性

协议解析的代码最容易在修改中引入回归。我习惯给handle_request写几个单元测试，覆盖正常读、正常写、越界读、不支持的功能码这四种情况。

import unittest class TestModbusHandler(unittest.TestCase): def test_read_holding_registers(self): # 构造读保持寄存器请求：事务ID=1，协议ID=0，长度=6，单元ID=1 # 功能码=0x03，起始地址=0，数量=2 request = struct.pack('>HHHBBHH', 1, 0, 6, 1, 0x03, 0, 2) resp = handle_request(request) # 检查响应的事务ID和功能码 trans_id, proto_id, length, unit_id = struct.unpack('>HHHB', resp[:7]) self.assertEqual(trans_id, 1) self.assertEqual(resp[7], 0x03) def test_illegal_address(self): # 请求超出范围的地址 request = struct.pack('>HHHBBHH', 2, 0, 6, 1, 0x03, 200, 2) resp = handle_request(request) self.assertEqual(resp[7], 0x83) # 异常功能码 self.assertEqual(resp[8], 0x02) # 非法数据地址 if __name__ == '__main__': unittest.main()

逻辑说明：用struct.pack手工构造请求字节，调用handle_request后检查响应的关键字段。这样每次改代码跑一遍测试，心里有底。

参数说明：struct.pack的格式字符串'>HHHBBHH'对应大端序下的字段排列，和 MBAP 头加 PDU 的结构一一对应。写测试时最容易错的是长度字段，记得数清楚。

5.4 一个我踩过的坑：字节序

Modbus 协议规定多字节字段用大端序，但有些设备厂商在寄存器内部存储 32 位浮点数时，会做字交换。比如一个浮点数占两个寄存器，高字和低字的顺序可能和标准相反。我在一个项目里调了整整一个下午，读上来的温度值一直是乱码，最后发现是服务端模拟器按标准大端存，而主站按字交换解析。后来我在模拟器里加了一个配置项，允许指定浮点数的字序，问题才解决。如果你要模拟的寄存器里存的是浮点数或 32 位整数，提前确认好字序，别等联调时才发现。

5.5 验证方法：用两个主站同时读写

最后分享一个验证模拟器稳定性的方法：开两个主站客户端，一个循环读，一个循环写，跑上半小时。观察是否有响应错乱、事务标识符不匹配、或者寄存器值被覆盖的情况。这个压力测试能暴露大部分并发问题。我一般用两个终端分别跑 pymodbus 脚本，一个读 0 到 10 号寄存器，一个随机写 0 到 10 号寄存器，看读到的值是否在合理范围内变化。如果出现读到的值突然变成 0 或者异常大，说明共享寄存器有竞态，需要加锁。

希望帮到你。

本文还有配套的精品资源，点击获取