基于Pymodbus监听多客户端保持寄存器变更的优化方案咨询

首先明确：Modbus TCP本身是主从架构，绝大多数常规自动化设备没有原生的主动变更通知机制，所以如果你的设备不支持自定义的主动上报功能（比如设备手册提到可触发事件推送数据），那还是得基于轮询优化，配合多线程/异步来提升多设备的监听性能。

一、先看有没有更优的变更通知方式 #

1. 优先检查设备主动上报能力
   部分高端工业设备会支持自定义功能码或事件触发的主动推送（比如寄存器值变化时，主动向你的服务器发送Modbus请求），如果设备有这个功能，直接配置开启即可，这是最优解，完全避免轮询带来的性能问题。但这不是标准Modbus功能，必须对应设备手册确认。
2. 优化轮询策略（仅当只能轮询时）
   • 不要重复建立连接：你现有代码里的while client.connect()逻辑有问题，每次循环都调用connect会反复建立TCP连接，应该保持长连接，仅在连接断开时重试重连，减少握手开销。
   • 调整轮询间隔：根据业务对实时性的要求设置合理间隔，比如数据变化不频繁的话，设为1-5秒，大幅降低请求量。
   • 批量读取寄存器：如果需要监听多个寄存器段，尽量一次读取所有目标寄存器，减少TCP交互次数。

二、多线程/异步是解决多设备性能问题的必要手段 #

单线程串行轮询10个设备时，每个设备的网络延迟、响应时间会累加，导致总耗时高、性能下降。用多线程或异步IO可以并行处理多个设备的轮询，大幅提升效率。

1. 多线程实现示例（基于你的代码修改）

每个设备对应独立线程，各自维护连接与轮询逻辑：

from pymodbus.client import ModbusTcpClient
import threading
import time

def monitor_device(ip, register_addr, register_count):
    client = ModbusTcpClient(ip)
    # 初始连接
    if not client.connect():
        print(f"连接失败：{ip}")
        return
    
    # 读取初始值
    try:
        result = client.read_holding_registers(register_addr, register_count)
        old_value = "".join([chr(code) for code in result.registers])
    except Exception as e:
        print(f"[{ip}] 初始读取失败：{e}")
        client.close()
        return
    
    while True:
        # 连接断开时重试
        if not client.is_socket_open():
            print(f"[{ip}] 重连中...")
            if not client.connect():
                time.sleep(2)
                continue
        
        try:
            result = client.read_holding_registers(register_addr, register_count)
            current_value = "".join([chr(code) for code in result.registers])
            
            if current_value != old_value:
                print(f"[{ip}] 值变更：{old_value} -> {current_value}")
                # 此处添加存储新值的逻辑，注意线程安全
                old_value = current_value
        except Exception as e:
            print(f"[{ip}] 读取错误：{e}")
        
        # 设置轮询间隔，示例为1秒
        time.sleep(1)
    
    client.close()

# 配置10个设备参数
devices = [
    ("192.168.5.1", 130, 12),
    ("192.168.5.2", 130, 12),
    # 补充剩余8个设备的参数
]

# 启动各设备监听线程
for dev in devices:
    thread = threading.Thread(target=monitor_device, args=dev, daemon=True)
    thread.start()

# 主线程保持运行
while True:
    time.sleep(3600)

2. 异步IO实现（更高效，适合大量设备）

用pymodbus的异步客户端配合asyncio，比多线程更节省资源，尤其设备数量较多时：

from pymodbus.client import AsyncModbusTcpClient
import asyncio

async def monitor_device(ip, register_addr, register_count):
    client = AsyncModbusTcpClient(ip)
    await client.connect()
    
    if not client.connected:
        print(f"连接失败：{ip}")
        return
    
    try:
        result = await client.read_holding_registers(register_addr, register_count)
        old_value = "".join([chr(code) for code in result.registers])
    except Exception as e:
        print(f"[{ip}] 初始读取失败：{e}")
        await client.close()
        return
    
    while True:
        if not client.connected:
            print(f"[{ip}] 重连中...")
            await client.connect()
            if not client.connected:
                await asyncio.sleep(2)
                continue
        
        try:
            result = await client.read_holding_registers(register_addr, register_count)
            current_value = "".join([chr(code) for code in result.registers])
            
            if current_value != old_value:
                print(f"[{ip}] 值变更：{old_value} -> {current_value}")
                # 存储新值，注意异步场景的协程安全
                old_value = current_value
        except Exception as e:
            print(f"[{ip}] 读取错误：{e}")
        
        await asyncio.sleep(1)
    
    await client.close()

async def main():
    devices = [
        ("192.168.5.1", 130, 12),
        ("192.168.5.2", 130, 12),
        # 补充剩余8个设备的参数
    ]
    tasks = [monitor_device(*dev) for dev in devices]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

注意事项 #

• 异常处理：必须添加异常捕获，避免单个设备的读取失败导致整个线程/协程崩溃。
• 数据存储安全：如果多个线程/协程同时写入同一存储（如数据库、文件），需使用锁（线程场景）或异步存储库（异步场景），避免数据冲突。
• 连接复用：始终保持长连接，不要每次读取都重新建立连接，减少TCP握手开销。

内容的提问来源于stack exchange，提问作者Hamidou