OPC UA节点恢复在线后如何获取数据变更通知并避免数据丢失？

核心思路：利用asyncua订阅的状态回调机制 #

asyncua的Subscription原生支持监控项的状态变更通知，无需自行轮询。通过自定义Handler监听监控项状态变化，既能实时知晓节点恢复情况，又能配合历史数据读取避免数据丢失。

具体实现方案 #

1. 自定义Subscription Handler，重写状态变更方法 #

在Handler中实现monitored_item_status_changed方法，该方法会在监控项状态（节点不可用/恢复可用）变化时自动触发：

class SubscriptionHandler:
    def __init__(self, client, subscription):
        self.client = client
        self.subscription = subscription
        self.unavailable_nodes = set()

    def datachange_notification(self, node, val, data):
        # 正常接收数据变更的逻辑
        print(f"节点 {node} 值更新：{val}")

    def monitored_item_status_changed(self, item_id, status, diagnostic):
        # 监控项状态变更回调
        node = self.subscription._monitored_items[item_id]
        if status.is_good():
            # 节点恢复可用
            print(f"节点 {node} 已恢复，状态：{status}")
            if node in self.unavailable_nodes:
                self.unavailable_nodes.remove(node)
                # 启动数据恢复任务
                asyncio.create_task(self.recover_missing_data(node))
        else:
            # 节点变为不可用
            print(f"节点 {node} 不可用，状态：{status}")
            self.unavailable_nodes.add(node)

    async def recover_missing_data(self, node):
        # 恢复后读取当前值或历史数据，填补缺失
        try:
            current_val = await node.read_value()
            print(f"恢复节点 {node} 当前值：{current_val}")
            # 若服务器开启历史存储，可读取不可用期间的历史数据
            # history_details = ua.ReadRawModifiedDetails()
            # history_vals = await node.read_history_values(history_details)
            # print(f"节点 {node} 历史数据：{history_vals}")
        except Exception as e:
            print(f"恢复节点 {node} 数据失败：{e}")

2. 初始化订阅时绑定自定义Handler #

修改订阅代码，传入自定义Handler实例：

import asyncio
import asyncua
from asyncua import ua

async def main():
    url = "opc.tcp://your-server-address:port"
    async with asyncua.Client(url=url) as client:
        handler = SubscriptionHandler(client, None)
        subscription = await client.create_subscription(50, handler)
        handler.subscription = subscription  # 将订阅实例关联到Handler
        nodes = [
            client.get_node('ns=3;i=1025'),
            client.get_node('ns=3;i=1026'),
            client.get_node('ns=3;i=1027')
        ]
        await subscription.subscribe_data_change(nodes)
        print("已订阅节点：", subscription._monitored_items)
        
        # 保持客户端运行
        await asyncio.Future()

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

3. 可选：历史数据补全 #

如果服务器配置了历史数据存储，在节点恢复时调用read_history_values方法，即可获取节点不可用期间的所有数据，完全避免丢失。

方案优势 #

• 实时感知：节点状态变化瞬间触发回调，无需等待轮询周期，不会错过恢复时机
• 无数据遗漏：恢复后可读取当前值或历史数据，覆盖不可用期间的缺失
• 资源高效：无需周期性发起读取请求，降低客户端与服务器的通信开销

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