使用Python freeopcua实现OPC UA订阅时如何访问数据

结论先行 #

你不需要额外调用node.get_value()获取变量值，这种主动拉取的方式确实会绕开你已经配置的订阅推送机制，违背了PubSub模式的设计初衷。

你只需要调整订阅回调处理器的实现，把推送过来的变更值存储到类属性中，主循环直接读取类属性即可，完整实现方案如下：

步骤1：改造SubHandler回调类，支持存储最新推送值 #

由于datachange_notification是freeopcua后台线程触发的回调，和主循环分属不同线程，需要加锁保证数据读写安全：

from threading import Lock
import time
from opcua import Client

class SubHandler(object):
    """
    客户端订阅回调处理器，接收服务端推送的事件/数据变更
    """
    def __init__(self):
        # 用字典存储所有节点的最新值，key为节点ID字符串，value为最新推送值
        self.latest_values = {}
        # 读写锁，避免多线程同时操作出现脏数据
        self.lock = Lock()

    def datachange_notification(self, node, val, data):
        node_id = node.nodeid.to_string()
        with self.lock:
            self.latest_values[node_id] = val
        # 实时性要求极高的逻辑可以直接在这里执行，无需等待主循环调度
    
    def event_notification(self, event):
        print("收到新事件", event)
    
    # 封装读方法，对外屏蔽锁逻辑
    def get_latest_value(self, node):
        node_id = node.nodeid.to_string()
        with self.lock:
            return self.latest_values.get(node_id, None)

步骤2：主循环直接读取回调类存储的推送值 #

if __name__ == "__main__":
    client = Client("opc.tcp://127.0.0.1:10000")
            
    try:
        client.connect()
        # 映射信息模型节点到Python对象，示例仅做参考
        node1 = client.get_node("ns=2;s=YourNodeID1")
        node2 = client.get_node("ns=2;s=YourNodeID2")
         
        handler = SubHandler()
        # 500为采样间隔，单位毫秒，服务端最快每500ms推送一次变更数据
        sub = client.create_subscription(500, handler)
        
        handle1 = sub.subscribe_data_change(node1)
        handle2 = sub.subscribe_data_change(node2)
        
        while True: 
            # 直接从回调处理器获取最新推送值，无需主动调用get_value拉取
            node1_val = handler.get_latest_value(node1)
            node2_val = handler.get_latest_value(node2)
            
            # 确认值已经被推送后再执行业务逻辑，避免空值报错
            if node1_val is not None and node2_val is not None:
                # 在这里执行你的业务逻辑
                # ...
                # 写变量逻辑保持不变，正常调用set_value即可
                # target_node.set_value(calculated_value)
            
            # 按需添加休眠，避免主循环占满CPU
            time.sleep(0.1)
    
    except Exception as e:
        print(f"运行异常：{e}")
        # 清理逻辑保持不变
        sub.unsubscribe(handle1)
        sub.unsubscribe(handle2)
        sub.delete()
        client.disconnect()

补充说明 #

• 主动调用node.get_value()本身不会破坏订阅机制的运行，订阅还是会正常收到推送，但这种主动轮询的方式会额外占用通信带宽，也失去了订阅模式“仅在数据变更时推送、无需重复轮询”的核心优势。
• 如果你的业务逻辑触发条件和节点变更强绑定，不需要固定周期执行，也可以直接把逻辑写在datachange_notification回调里，延迟会更低。

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