opcua-asyncio多设备节点订阅问题：多连接下Handler未触发

问题根源 #

opcua-asyncio 是异步驱动的库，若用同步逻辑循环创建连接、或错误放置阻塞代码（比如单独的 while True），会打断事件循环调度，导致订阅回调无法触发。每个设备的连接、订阅都需要作为独立异步任务，由事件循环统一管理。

具体实现方案 #

1. 带设备标识的订阅处理器

给处理器添加设备ID，方便区分不同设备的节点更新：

from asyncua import Client, ua
from asyncua.common.subscription import SubscriptionHandler

class DeviceSubscriptionHandler(SubscriptionHandler):
    def __init__(self, device_id):
        super().__init__()
        self.device_id = device_id

    def datachange_notification(self, node, val, data):
        print(f"设备{self.device_id} | 节点{node} | 更新值: {val}")

2. 封装单设备连接订阅异步函数

每个设备用独立的 Client 和 Subscription，内部通过 client.run_forever() 维持连接，无需手动写 while True：

async def connect_and_subscribe(device_url, device_id, node_ids):
    client = Client(url=device_url)
    try:
        await client.connect()
        print(f"设备{device_id}已连接: {device_url}")

        # 创建订阅及处理器
        handler = DeviceSubscriptionHandler(device_id)
        subscription = await client.create_subscription(500, handler)  # 500ms 发布间隔

        # 批量订阅节点
        target_nodes = [await client.get_node(nid) for nid in node_ids]
        await subscription.subscribe_data_change(target_nodes)

        # 阻塞维持连接，直到主动断开
        await client.run_forever()
    finally:
        await client.disconnect()
        print(f"设备{device_id}已断开连接")

3. 主函数并发管理多设备任务

用 asyncio.gather 让事件循环同时处理所有设备的连接与订阅任务：

import asyncio

async def main():
    # 设备配置清单，按需添加更多设备
    devices = [
        {"url": "opc.tcp://device1:4840", "id": 1, "nodes": ["ns=2;i=1", "ns=2;i=2"]},
        {"url": "opc.tcp://device2:4840", "id": 2, "nodes": ["ns=2;i=3", "ns=2;i=4"]},
    ]

    # 创建所有设备的异步任务
    tasks = [
        asyncio.create_task(
            connect_and_subscribe(dev["url"], dev["id"], dev["nodes"])
        )
        for dev in devices
    ]

    # 等待所有任务执行（手动中断则退出）
    await asyncio.gather(*tasks)

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

关键注意点 #

• 禁止在循环内单独写 while True，这会阻塞事件循环，导致后续设备的连接、订阅无法初始化。client.run_forever() 已封装了维持连接的逻辑，且不会阻塞事件循环的其他任务。
• 每个设备使用独立的 Client 和 Handler，避免共享资源引发的冲突。
• 所有异步任务必须由事件循环调度，asyncio.create_task 和 asyncio.gather 是并发管理多设备的核心。

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