本文详解 PyModbus TCP 服务器中 StartTcpServer 的阻塞特性问题，提供非阻塞更新数据的正确方案——使用线程安全的数据存储与后台更新线程，并给出可运行的完整服务端/客户端示例。
本文详解 pymodbus tcp 服务器中 `starttcpserver` 的阻塞特性问题，提供非阻塞更新数据的正确方案——使用线程安全的数据存储与后台更新线程，并给出可运行的完整服务端/客户端示例。
在使用 PyModbus 构建 Modbus TCP 服务器时，一个常见误区是误以为 StartTcpServer() 会“启动后返回”，从而在其后直接编写数据更新逻辑（如轮询修改寄存器值）。但实际上，StartTcpServer() 是一个完全阻塞式调用——它会独占主线程，进入事件循环监听客户端连接与请求，永远不会主动返回，因此其后的代码（如 update_data() 循环）根本不会执行。
你遇到的 [0], [0], ... 读取结果，正是因为服务端初始化后未做任何更新，所有寄存器始终维持初始值 0；而将 block.setValues() 放在 StartTcpServer() 之前虽能写入一次，但无法实现“动态更新”，自然也无法验证读取逻辑是否真正生效。
✅ 正确解法：使用线程安全的数据存储 + 后台更新线程
PyModbus 的 ModbusSequentialDataBlock 本身不是线程安全的，但在单个 ModbusSlaveContext 中，只要确保所有对 block.setValues() 的调用都发生在同一上下文（且不与服务器内部读操作发生竞态），配合合理同步即可安全使用。更稳妥的做法是借助 threading.Lock 或改用 ModbusServerContext 的 getValues/setValues 接口（它们已内置轻量同步）。
以下是修复后的完整、可运行的服务端代码（支持后台定时更新）：
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus.device import ModbusDeviceIdentification
import threading
import random
import time
# 初始化数据块：100 个 holding register（地址 0–99）
block = ModbusSequentialDataBlock(0, [0] * 100)
store = ModbusSlaveContext(
di=block, # Discrete Inputs
co=block, # Coils
hr=block, # Holding Registers ← 我们主要读写此处
ir=block # Input Registers
)
context = ModbusServerContext(slaves=store, single=True)
# 设备标识（可选，增强兼容性）
identity = ModbusDeviceIdentification()
identity.VendorName = 'FI'
identity.ProductCode = 'S1'
identity.VendorUrl = 'http://github.com/riptideio/pymodbus/'
identity.ProductName = 'PyModbus Dynamic Server'
identity.ModelName = 'S1'
identity.MajorMinorRevision = '1.0'
# ? 关键：定义线程安全的数据更新函数
def update_registers_periodically():
while True:
# 模拟动态数据：更新前10个 holding register
new_values = [random.randint(0, 255) for _ in range(10)]
try:
# ✅ 安全写入：通过 context 调用 setValues（推荐方式）
store.setValues(3, 0, new_values) # 3=HR, 0=start_addr, new_values=list
print(f"[更新] HR[0-9] → {new_values}")
except Exception as e:
print(f"写入失败: {e}")
time.sleep(5)
if __name__ == "__main__":
# ? 启动后台更新线程（非阻塞！）
updater_thread = threading.Thread(target=update_registers_periodically, daemon=True)
updater_thread.start()
print("✅ Modbus TCP 服务器启动中... (端口 5020)")
print("? 数据已启用每 5 秒自动更新，客户端可随时读取")
# ⚠️ StartTcpServer 阻塞在此处 —— 但 updater 已在后台运行
StartTcpServer(context, identity=identity, address=("localhost", 5020))? 客户端注意事项（关键修正）：
你的原始客户端代码中 client.read_holding_registers(i) 缺少 count= 参数，默认只读 1 个寄存器，但调用方式易出错。应统一使用：
result = client.read_holding_registers(address=i, count=1) # 或批量读取（更高效）： result = client.read_holding_registers(address=0, count=10)
✅ 推荐优化版客户端（含错误处理与批量读取）：
from pymodbus.client import ModbusTcpClient
import time
client = ModbusTcpClient('localhost', port=5020)
if client.connect():
print("✅ 客户端已连接至 Modbus 服务器")
try:
while True:
# 批量读取 HR 地址 0–9（共 10 个）
result = client.read_holding_registers(address=0, count=10)
if result.isError():
print("❌ 读取错误:", result)
else:
print("? 当前寄存器值:", result.registers)
time.sleep(5)
finally:
client.close()
else:
print("❌ 无法连接到服务器，请检查服务端是否运行")⚠️ 重要提醒：
- 不要尝试在
StartTcpServer()后写任何逻辑——它永不返回； - 动态更新必须在独立线程（或异步任务）中完成；
- 使用
store.setValues(3, addr, values)比直接操作block.setValues()更符合 PyModbus 最佳实践； - 客户端务必检查
result.isError()，Modbus 异常（如非法地址）不会抛出 Python 异常，而是返回错误响应对象； - 确保防火墙/网络允许本地
5020端口通信。
掌握这一模式后，你即可无缝扩展为传感器模拟、PLC 数据桥接、HMI 数据源等真实工业场景。
