Node-RED中Python编写的OPC Client运行1小时后报错求助

问题背景 #

用Python编写的OPC UA客户端在Node-RED中运行，负责将传感器数据上传至Thingsboard，运行约1小时后出现连接相关报错。

原代码 #

from opcua import Client
import paho.mqtt.client as mqtt
import time
import json

url = "opc.tcp://D201002809747:4840"
client = Client(url)
client.connect()
print ("已连接OPC服务器")

iot_hub = "iothub.toyota.co.id"
port = 1883  # MQTT默认端口，原代码留空会导致连接失败
username = "MdduLiSzC787aY8DrYNf"
password = ""
topic = "v1/devices/me/telemetry"

iot_hub_client = mqtt.Client()
iot_hub_client.username_pw_set(username, password)
iot_hub_client.connect(iot_hub, port)
print ("已连接IoT Hub")

data = dict()
while True:
    el = client.get_node("ns=2;s=Device1.DATA.MOTOR_CV3_ALONGATION")
    elongation = el.get_value()
    print (elongation)

    data['Elongation'] = int(elongation)
    data_out = json.dumps(data)
    iot_hub_client.publish(topic, data_out, 0)
    
    time.sleep(2)

注：原代码存在两处基础问题：while True循环内代码未缩进（语法错误）、MQTT端口留空，上述代码已先行修正。

错误信息（翻译后） #

退出代码：1
回溯（最近调用最后）：
  文件 "C:\Users\admin\coba.py", 第25行, in <module>
    elongation = el.get_value()
                ^^^^^^^^^^^^^^
  文件 "C:\Users\admin\AppData\Roaming\Python\Python311\site-packages\opcua\common\node.py", 第155行, in get_value
    result = self.get_data_value()
             ^^^^^^^^^^^^^^^^^^^^^
  文件 "C:\Users\admin\AppData\Roaming\Python\Python311\site-packages\opcua\common\node.py", 第164行, in get_data_value
    return self.get_attribute(ua.AttributeIds.Value)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  文件 "C:\Users\admin\AppData\Roaming\Python\Python311\site-packages\opcua\common\node.py", 第275行, in get_attribute
    result = self.server.read(params)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  文件 "C:\Users\admin\AppData\Roaming\Python\Python311\site-packages\opcua\client\ua_client.py", 第347行, in read
    data = self._uasocket.send_request(request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  文件...

问题原因分析 #

1. OPC UA连接无重连机制：长时间运行中，OPC服务器可能因超时、网络波动主动断开连接，原代码未处理连接丢失情况，一旦断开就会抛出请求失败错误。
2. MQTT连接未做保活：MQTT连接同样可能因网络问题断开，原代码无重连逻辑，会导致数据无法上传。
3. 缺少异常捕获：运行过程中出现的网络异常、节点读取异常未被捕获，直接导致程序崩溃。

修复后的代码 #

from opcua import Client, ua
import paho.mqtt.client as mqtt
import time
import json

# OPC UA配置
OPC_URL = "opc.tcp://D201002809747:4840"
NODE_ID = "ns=2;s=Device1.DATA.MOTOR_CV3_ALONGATION"

# MQTT配置
IOT_HUB = "iothub.toyota.co.id"
MQTT_PORT = 1883
USERNAME = "MdduLiSzC787aY8DrYNf"
PASSWORD = ""
TOPIC = "v1/devices/me/telemetry"

def connect_opc():
    """创建并连接OPC客户端"""
    try:
        opc_client = Client(OPC_URL)
        opc_client.connect()
        print("已重新连接OPC服务器")
        return opc_client
    except Exception as e:
        print(f"OPC连接失败: {str(e)}")
        return None

def connect_mqtt():
    """创建并连接MQTT客户端"""
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("已重新连接IoT Hub")
        else:
            print(f"MQTT连接失败，错误码: {rc}")
    
    mqtt_client = mqtt.Client()
    mqtt_client.username_pw_set(USERNAME, PASSWORD)
    mqtt_client.on_connect = on_connect
    # 开启自动重连，设置重连延迟范围
    mqtt_client.reconnect_delay_set(min_delay=1, max_delay=60)
    try:
        mqtt_client.connect(IOT_HUB, MQTT_PORT, keepalive=60)
        mqtt_client.loop_start()
        return mqtt_client
    except Exception as e:
        print(f"MQTT连接初始化失败: {str(e)}")
        return None

# 初始化连接
opc_client = connect_opc()
mqtt_client = connect_mqtt()

data = dict()
while True:
    # 检查并重建OPC连接
    if not opc_client or not opc_client.is_connected():
        opc_client = connect_opc()
        time.sleep(5)
        continue
    
    # 检查并重建MQTT连接
    if not mqtt_client or not mqtt_client.is_connected():
        mqtt_client = connect_mqtt()
        time.sleep(5)
        continue
    
    try:
        # 读取OPC节点值
        el = opc_client.get_node(NODE_ID)
        elongation = el.get_value()
        print(f"读取到伸长值: {elongation}")
        
        # 上传数据到Thingsboard
        data['Elongation'] = int(elongation)
        data_out = json.dumps(data)
        mqtt_client.publish(TOPIC, data_out, qos=0)
        
    except ua.UaError as e:
        print(f"OPC数据读取错误: {str(e)}")
        opc_client = None  # 标记连接失效，触发下一次重连
    except Exception as e:
        print(f"其他运行错误: {str(e)}")
    
    time.sleep(2)

关键改进点 #

• OPC连接重连逻辑：每次循环检查连接状态，断开时自动重建。
• MQTT自动重连：开启MQTT客户端的自动重连机制，设置保活时间，避免连接丢失。
• 异常捕获处理：捕获OPC UA专属错误和通用异常，程序不会因单次错误直接崩溃，而是尝试恢复。
• 代码结构优化：将连接逻辑封装成函数，提升可读性和维护性。

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