如何通过PyModbusTCP高效批量读取非连续保持寄存器？

嗨，我来帮你搞定这个高效读取非连续寄存器的问题！你现在逐个读取的方式确实会因为多次网络请求拖慢速度，毕竟每秒都要读，开销不小。下面给你几个实用的解决方案：

最优方案：合并连续区间读取（推荐） #

Modbus协议的读保持寄存器命令（功能码0x03）本身只支持读取连续的寄存器区间，所以最高效的方式就是把你需要的非连续寄存器，包裹在一个最小的连续区间里一次性读取，之后再从中提取你需要的对应值。这样只需要一次网络请求，比多次请求快太多了。

举个你的例子，你要读的寄存器是[32260, 32262, 32264]，最小的起始寄存器是32260，最大的结束寄存器是32264，中间一共包含5个连续寄存器（32260到32264），我们一次性读取这5个，再提取索引0、2、4对应的值即可：

import time
from pymodbus.client import ModbusClient

registers = [32260, 32262, 32264]
host = "你的设备IP"
port = 502  # 默认Modbus TCP端口
slave_id = 1  # 你的设备Unit ID

client = ModbusClient(host=host, port=port, unit_id=slave_id)
# 先确保连接成功
if not client.connect():
    print("连接设备失败！")
    exit()

while True:
    try:
        # 计算需要读取的连续区间范围
        start_reg = min(registers)
        end_reg = max(registers)
        reg_count = end_reg - start_reg + 1
        
        # 一次性读取整个区间的寄存器
        all_reg_values = client.read_holding_registers(start_reg, reg_count)
        
        if all_reg_values and not all_reg_values.isError():
            # 提取目标寄存器对应的值
            target_values = []
            for reg in registers:
                # 计算当前寄存器在返回列表中的索引
                idx = reg - start_reg
                target_values.append(all_reg_values.registers[idx])
            print("目标寄存器值:", target_values)
        else:
            print("读取失败:", all_reg_values)
        
        time.sleep(1)
    except Exception as e:
        print(f"读取出错: {str(e)}")
        time.sleep(1)

备选方案：多线程并行读取（适用于寄存器极度分散的情况） #

如果你的非连续寄存器分布得非常零散，没法用连续区间包裹（比如跨度特别大，读取多余寄存器的开销反而更大），那可以用多线程并行读取，同时发起多个请求，减少整体等待时间。

注意：要确保你的设备支持同时处理多个Modbus请求，另外每个线程最好单独创建客户端连接，避免线程安全问题：

import time
import threading
from queue import Queue
from pymodbus.client import ModbusClient

def read_single_register(reg_addr, result_queue, host, port, slave_id):
    # 每个线程单独创建客户端
    client = ModbusClient(host=host, port=port, unit_id=slave_id)
    try:
        if client.connect():
            response = client.read_holding_registers(reg_addr, 2)
            if response and not response.isError():
                result_queue.put((reg_addr, response.registers))
            else:
                result_queue.put((reg_addr, None))
    except Exception as e:
        result_queue.put((reg_addr, f"错误: {str(e)}"))
    finally:
        client.close()

registers = [32260, 32262, 32264]
host = "你的设备IP"
port = 502
slave_id = 1

while True:
    result_queue = Queue()
    threads = []
    
    # 为每个寄存器创建读取线程
    for reg in registers:
        thread = threading.Thread(
            target=read_single_register,
            args=(reg, result_queue, host, port, slave_id)
        )
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()
    
    # 整理结果
    final_results = {}
    while not result_queue.empty():
        reg_addr, value = result_queue.get()
        final_results[reg_addr] = value
    
    print("批量读取结果:", final_results)
    time.sleep(1)

额外说明 #

• PyModbusTCP的read_holding_registers方法本身不支持直接传入非连续寄存器列表，这是因为Modbus协议的限制——功能码0x03只能读取连续寄存器。
• 优先用第一种合并区间的方法，因为网络请求的开销远大于读取几个多余寄存器的计算开销，一次请求的效率远高于多次请求。

备注：内容来源于stack exchange，提问作者dpatrik0630