4. OPC UA 服务器数据订阅
任务目标
编写一个OPC UA客户端程序，连接到指定的OPC UA服务器，订阅服务器上某个节点的数据变化，并在数据变化时实时打印新值。本案例使用开源Python库asyncua实现，服务器地址与节点ID作为可配置参数。
环境依赖
- Python 3.8+
- 安装依赖：
pip install asyncua cryptography
关键完整代码
import asyncio
import sys
from asyncua import Client, ua
# 配置参数
SERVER_URL = "opc.tcp://localhost:4840" # OPC UA服务器地址
NODE_ID = "ns=2;i=5" # 要订阅的节点ID（示例：命名空间2，标识符5）
async def subscribe_data_change():
"""连接服务器并订阅指定节点的数据变化"""
client = Client(SERVER_URL)
try:
# 建立连接
await client.connect()
print(f"已连接到服务器: {SERVER_URL}")
# 获取目标节点
node = client.get_node(NODE_ID)
# 验证节点存在并可读
value = await node.read_value()
print(f"当前节点值: {value}")
# 创建订阅
subscription = await client.create_subscription(100, handler) # 100ms发布间隔
# 订阅节点数据变化
handle = await subscription.subscribe_data_change(node)
print(f"已订阅节点 {NODE_ID} 的数据变化，等待数据变更...")
# 持续运行，等待数据变化回调
while True:
await asyncio.sleep(1)
except Exception as e:
print(f"发生错误: {e}")
sys.exit(1)
finally:
# 断开连接
await client.disconnect()
print("已断开连接")
class handler:
"""数据变化回调处理类"""
def datachange_notification(self, node, value, data):
"""数据变化时被调用的回调方法"""
print(f"节点 {node} 数据变化: {value.value} (时间: {data.monitored_item_value.timestamp})")
if __name__ == "__main__":
asyncio.run(subscribe_data_change())
运行与验证
- 准备服务器：确保OPC UA服务器运行在
localhost:4840，且节点ns=2;i=5存在并允许订阅。若使用Prosys Simulation Server等模拟服务器，可直接创建该节点。 - 运行客户端：
python opcua_subscriber.py - 验证通过条件：
- 程序输出“已连接到服务器”和当前节点值，说明连接与读取成功；
- 在服务器端修改该节点的值（如通过服务器管理界面或另一个写入客户端），程序应打印出包含新值和时间戳的数据变化通知；
- 连续修改多次，每次变化均能实时打印，说明订阅机制正常工作。
常见问题
- 连接失败：检查服务器地址、端口是否正确，服务器是否允许匿名访问或需要配置安全策略。
- 节点不存在：确认节点ID格式正确（
ns=命名空间索引;i=标识符），可通过UA Expert等工具浏览服务器节点树确认。 - 收不到数据变化：确认服务器端节点确实发生了值变化（而非仅写入相同值），且订阅发布间隔（代码中为100ms）设置合理。