3. OPC UA 服务器数据订阅

任务目标

本案例将实现一个OPC UA客户端，连接PLC模拟器或真实PLC的OPC UA服务器，订阅指定的数据节点。当服务器端数据发生变化时，客户端能够实时收到通知并打印最新值。通过本案例，掌握OPC UA订阅机制的核心配置流程。

环境依赖

使用Python与asyncua库（纯Python实现，无需额外C++依赖）。安装命令：

pip install asyncua

完整实现代码

以下代码创建OPC UA客户端，连接服务器，订阅ns=2;i=2（模拟器中的随机变化数据节点），持续监听数据变化：

import asyncio
from asyncua import Client

SERVER_URL = "opc.tcp://localhost:4840"
NODE_ID = "ns=2;i=2"  # 模拟器中的动态数据节点

async def subscribe_data():
    client = Client(SERVER_URL)
    try:
        await client.connect()
        print(f"已连接: {SERVER_URL}")

        # 获取目标节点
        node = client.get_node(NODE_ID)
        # 读取初始值（验证连接与节点有效性）
        initial_value = await node.read_value()
        print(f"初始值: {initial_value}")

        # 创建订阅
        subscription = await client.create_subscription(100, handler)  # 100ms轮询间隔
        # 订阅节点数据变化
        handle = await subscription.subscribe_data_change(node)
        print(f"订阅成功, handle={handle}, 等待数据变化... (Ctrl+C退出)")

        # 持续运行接收回调
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        print("\n用户中断")
    finally:
        await client.disconnect()
        print("已断开连接")

class handler:
    """数据变化回调类"""
    def datachange_notification(self, node, value, data):
        print(f"[{node}] 新值: {value}")

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

运行与验证

第一步：启动OPC UA服务器（模拟器）。使用asyncua自带的示例服务器快速验证：

python -m asyncua.server --port 4840

该命令启动一个本地服务器，包含ns=2;i=2节点，其值每500ms随机变化一次。

第二步：运行客户端脚本。将上述代码保存为opcua_subscriber.py，执行：

python opcua_subscriber.py

验证通过条件：

• 控制台输出已连接与初始值，说明连接与节点读取正常；
• 输出订阅成功后，持续打印[Node(...)] 新值: X，且数值随时间变化（非固定值），说明订阅生效；
• 若服务器端数据停止变化（如模拟器暂停），则不再有新值输出，证明是实时订阅而非轮询。

关键说明

• 订阅间隔：create_subscription(100, handler)中100为发布间隔（毫秒），即服务器每100ms检查一次变化并推送。实际延迟取决于服务器端发布周期与网络。
• 回调类：handler类必须实现datachange_notification方法，否则无法接收通知。若需处理事件（非数据变化），需额外实现event_notification。
• 节点标识：ns=2;i=2是模拟器预置节点。若连接真实PLC，需根据其地址空间文档修改NODE_ID，例如ns=3;s=Channel1.Device1.Tag1。
• 断开清理：程序退出时调用disconnect()释放订阅与连接资源，避免服务器端残留无效会话。

若需同时订阅多个节点，可多次调用subscription.subscribe_data_change(node)，每个节点返回独立handle，回调中通过node参数区分来源。