使用pymodbus的read_holding_registers遇错误端口程序冻结，无法捕获异常求助

核心原因 #

连接蓝牙虚拟串口这类无实际Modbus设备响应的端口时，ModbusSerialClient.connect()可能返回成功（因为串口本身能被打开），但后续read_holding_registers()会卡在底层串口读写操作中——由于没有设备返回响应，且底层串口驱动未触发超时错误，导致程序冻结，常规ModbusException和通用异常无法捕获这种阻塞。

解决方案 #

1. 提前验证端口有效性 #

连接前先检查端口是否存在且可正常访问，避免无效连接：

import serial.tools.list_ports

def is_port_valid(port):
    ports = [p.device for p in serial.tools.list_ports.comports()]
    if port not in ports:
        return False
    try:
        ser = serial.Serial(port, baudrate=9600, timeout=1)
        ser.close()
        return True
    except serial.SerialException:
        return False

在read_holding_registers函数开头加入：

if not is_port_valid(port):
    logger.error(f"Port {port} is invalid or unavailable.")
    return None

2. 为读操作添加强制超时 #

用线程给读操作设置最大执行时间，避免无限阻塞：

import threading
import inspect
import ctypes

def terminate_thread(thread):
    if not thread.is_alive():
        return
    exc = ctypes.py_object(SystemExit)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), exc)
    if res > 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), None)

def read_with_timeout(client, start_addr, count, slave_id, timeout=2):
    result = None
    def target():
        nonlocal result
        result = client.read_holding_registers(address=start_addr, count=count, slave=slave_id)
    
    thread = threading.Thread(target=target)
    thread.start()
    thread.join(timeout)
    
    if thread.is_alive():
        terminate_thread(thread)
        logger.error("Read operation timed out.")
        return None
    return result

替换原代码中读操作的部分：

response = read_with_timeout(client, start_address, count, slave_id, timeout=2)
if response is None:
    return None

3. 捕获底层串口异常 #

导入serial.SerialException并添加捕获逻辑，处理串口层面的错误：

from serial import SerialException

# 在try-except块中新增
except SerialException as e:
    logger.error(f"Serial Port Error: {e}")
    return None

4. 启用严格Modbus协议模式 #

初始化客户端时添加strict=True，强制遵循协议规范，减少无响应时的阻塞概率：

client = ModbusSerialClient(
    port=port,
    baudrate=9600,
    stopbits=1,
    bytesize=8,
    parity='N',
    timeout=2,
    strict=True  # 启用严格模式
)

修改后的完整代码 #

from pymodbus.client import ModbusSerialClient
from pymodbus.exceptions import ModbusException
from serial import SerialException
import serial.tools.list_ports
import logging
import threading
import inspect
import ctypes

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def is_port_valid(port):
    ports = [p.device for p in serial.tools.list_ports.comports()]
    if port not in ports:
        return False
    try:
        ser = serial.Serial(port, baudrate=9600, timeout=1)
        ser.close()
        return True
    except SerialException:
        return False

def terminate_thread(thread):
    if not thread.is_alive():
        return
    exc = ctypes.py_object(SystemExit)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), exc)
    if res > 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), None)

def read_with_timeout(client, start_addr, count, slave_id, timeout=2):
    result = None
    def target():
        nonlocal result
        result = client.read_holding_registers(address=start_addr, count=count, slave=slave_id)
    
    thread = threading.Thread(target=target)
    thread.start()
    thread.join(timeout)
    
    if thread.is_alive():
        terminate_thread(thread)
        logger.error("Read operation timed out.")
        return None
    return result

def read_holding_registers(port, slave_id, start_address, count):
    if not is_port_valid(port):
        logger.error(f"Port {port} is invalid or unavailable.")
        return None

    client = ModbusSerialClient(
        port=port,
        baudrate=9600,
        stopbits=1,
        bytesize=8,
        parity='N',
        timeout=2,
        strict=True
    )

    try:
        if not client.connect():
            logger.error("Failed to connect to the device.")
            return None

        response = read_with_timeout(client, start_address, count, slave_id, timeout=2)
        if response is None:
            return None

        if response.isError():
            logger.error(f"Modbus Error: {response}")
            return None

        logger.info(f"Successfully read {len(response.registers)} registers.")
        return response.registers

    except ModbusException as e:
        logger.error(f"Modbus Exception: {e}")
        return None
    except SerialException as e:
        logger.error(f"Serial Port Error: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected Error: {e}")
        return None
    finally:
        client.close()

if __name__ == "__main__":
    PORT = 'COM7'
    SLAVE_ID = 1
    START_ADDR = 0
    REG_COUNT = 10

    print(f"Reading {REG_COUNT} Holding Registers starting from address {START_ADDR}...")
    registers = read_holding_registers(PORT, SLAVE_ID, START_ADDR, REG_COUNT)

    if registers is not None:
        for i, reg in enumerate(registers):
            print(f"Register {START_ADDR + i}: {reg} (0x{reg:04X})")
    else:
        print("Failed to read registers. Check connection and parameters.")

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