4. OPC UA 服务器数据订阅

任务目标

本案例将实现一个最小可运行的 OPC UA 客户端，连接本地 OPC UA 服务器，订阅服务器上某个节点的数据变化，并实时打印最新值。通过本案例，掌握 OPC UA 订阅机制的核心调用流程。

环境与依赖

使用 Python 与 asyncua 库（基于 asyncio）。安装依赖：

pip install asyncua

需要本地有一个可访问的 OPC UA 服务器。若没有现成服务器，可使用 asyncua 自带的示例服务器（见下文验证步骤）。

关键完整代码

import asyncio
import logging
from asyncua import Client

logging.basicConfig(level=logging.INFO)

# OPC UA 服务器地址（根据实际服务器修改）
SERVER_URL = "opc.tcp://localhost:4840"
# 要订阅的节点 ID（示例：ns=2;i=1 为服务器模拟量节点）
NODE_ID = "ns=2;i=1"

async def subscribe_data():
    client = Client(SERVER_URL)
    try:
        await client.connect()
        logging.info("已连接服务器: %s", SERVER_URL)

        # 获取目标节点
        node = client.get_node(NODE_ID)

        # 创建订阅（周期 1000ms）
        subscription = await client.create_subscription(1000, SubHandler())
        await subscription.subscribe_data_change(node)
        logging.info("已订阅节点: %s", NODE_ID)

        # 持续运行，等待数据变化回调
        while True:
            await asyncio.sleep(1)
    finally:
        await client.disconnect()

class SubHandler:
    """订阅数据变化回调处理类"""
    def datachange_notification(self, node, val, data):
        logging.info("节点 %s 新值: %s", node, val)

    def event_notification(self, event):
        logging.info("事件通知: %s", event)

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

运行与验证

第一步：启动 OPC UA 服务器（若已有服务器可跳过）。在另一个终端运行：

python -m asyncua.server

该命令会启动一个默认服务器，监听 opc.tcp://localhost:4840，并包含一个模拟量节点 ns=2;i=1，其值会周期性变化。

第二步：运行订阅客户端

python opcua_subscriber.py

验证通过条件：

1. 客户端日志显示 已连接服务器 与 已订阅节点，无报错。
2. 持续输出类似 节点 NodeId(...) 新值: 42.5 的日志，且数值随时间变化，说明订阅生效并实时接收数据。

常见问题

• 连接失败：确认服务器已启动，且 SERVER_URL 与服务器实际地址一致。
• 节点不存在：若使用自定义服务器，需将 NODE_ID 改为实际存在的节点 ID。可通过 asyncua 的浏览功能查询节点。
• 收不到数据变化：确认所订阅节点确实在周期性变化；若节点值不变，订阅不会触发回调。