如何在 PyModbus 中正确读取服务器端写入的寄存器值-Python教程-PHP中文网

如何在 PyModbus 中正确读取服务器端写入的寄存器值

本文详解 PyModbus TCP 服务端 StartTcpServer 的阻塞特性及其对数据更新的影响，提供非阻塞更新方案（如线程/异步机制）与客户端读取的最佳实践，确保客户端能实时读取服务端动态更新的寄存器值。

本文详解 pymodbus tcp 服务端 `starttcpserver` 的阻塞特性及其对数据更新的影响，提供非阻塞更新方案（如线程/异步机制）与客户端读取的最佳实践，确保客户端能实时读取服务端动态更新的寄存器值。

PyModbus 的 StartTcpServer 是一个完全阻塞式调用——一旦执行，程序控制权即交由内置事件循环接管，后续代码（如你的 update_data() 循环）将永不执行。这正是你观察到“block.setValues() 在 StartTcpServer() 之后不生效”以及客户端始终读到 [0] 的根本原因：服务端数据从未被更新。

✅ 正确做法：在服务端上下文内实现异步/并发更新

PyModbus v3.5.0+ 支持通过 custom_functions 或更推荐的方式——使用线程安全的数据存储 + 后台线程更新。以下是修复后的完整服务端示例：

from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
from pymodbus.device import ModbusDeviceIdentification
import random
import threading
import time

# 初始化数据块（100个保持寄存器，地址0–99）
block = ModbusSequentialDataBlock(0, [0] * 100)
store = ModbusSlaveContext(
    di=block, co=block, hr=block, ir=block  # 所有区域共享同一数据块
)
context = ModbusServerContext(slaves=store, single=True)

identity = ModbusDeviceIdentification()
identity.VendorName = 'FI'
identity.ProductCode = 'S1'
identity.VendorUrl = 'http://github.com/riptideio/pymodbus/'
identity.ProductName = 'Mein Server'
identity.ModelName = 'S1'
identity.MajorMinorRevision = '1.0'

# ? 后台线程：安全更新寄存器（线程安全，因ModbusSequentialDataBlock内部已加锁）
def update_registers():
    while True:
        # 更新前10个保持寄存器（地址0–9）
        values = [random.randint(0, 255) for _ in range(10)]
        block.setValues(0, values)  # 地址0开始写入10个值
        print(f"[更新] HR[0-9] → {values}")
        time.sleep(5)

if __name__ == '__main__':
    # ? 关键：启动更新线程（必须在 StartTcpServer 前！）
    updater = threading.Thread(target=update_registers, daemon=True)
    updater.start()

    print("✅ Modbus TCP Server 启动中... (端口 5020)")
    # StartTcpServer 将在此处阻塞并运行服务
    StartTcpServer(context, identity=identity, address=("localhost", 5020))

⚠️ 注意事项：

daemon=True 确保主线程退出时后台线程自动终止；

block.setValues() 是线程安全的（PyModbus 内部已同步），无需额外加锁；

切勿在 StartTcpServer() 后写任何逻辑——它永不返回。

✅ 客户端读取优化（推荐）

你的客户端代码基本正确，但可增强健壮性与可读性：

from pymodbus.client import ModbusTcpClient
import time

client = ModbusTcpClient('localhost', port=5020)

if client.connect():
    print("✅ 客户端已连接至 Modbus 服务器")
    try:
        while True:
            # 一次性读取10个连续寄存器（更高效）
            result = client.read_holding_registers(address=0, count=10, slave=1)
            if result.isError():
                print(f"❌ 读取错误: {result}")
            else:
                print(f"? 当前值 HR[0–9]: {result.registers}")
            time.sleep(3)
    except KeyboardInterrupt:
        print("\n⏹️  用户中断读取")
    finally:
        client.close()
else:
    print("❌ 无法连接到服务器，请检查服务端是否运行")

✅ 验证与调试建议

使用标准工具验证（如 modbus-cli 或 QModMaster）排除代码逻辑干扰；
在服务端 update_registers() 中添加 print(block.values[:10]) 确认写入生效；
检查防火墙/端口占用：netstat -ano | findstr :5020（Windows）或 lsof -i :5020（macOS/Linux）；
若需更高实时性，可考虑 pymodbus 的异步服务端（StartAsyncTcpServer + asyncio），但需 Python ≥ 3.7 且逻辑重构。