OPCUA异步客户端批量取标签任务创建耗时过长，求优化方案

问题描述 #

我尝试通过路径ns=2;s=TEST.PLC1从OPC UA服务器获取超过19000个标签。目前的实现是先构建Tags_List字典（格式为"TEST1":"ns=2;s=TEST1"），耗时0.5秒；接着创建任务列表tasks，添加每个标签的取值任务，通过await asyncio.gather(*tasks)并发执行，这一步耗时3.5秒。想请教：

• 当前操作存在哪些问题？
• 如何缩短tasks的创建时间？
• 最终目标是在1秒内获取50000个标签的值，该如何实现？

当前代码实现 #

import asyncio
import sys
import aiopg
from asyncua import Client as ClientASY
import datetime

# asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# asyncio.Semaphore(value=9999)

url = "opc.tcp://127.0.0.1:49320"

async def create_tags_dict(path, client):
    tag_list = dict()
    node = client.get_node(path)
    variables = await node.get_variables()
    for i in variables:
        tag_list[i.nodeid.Identifier] = i
    return tag_list

async def get_value_from_dict(address, client):
    value = await client.get_node(address).get_value()
    return value

async def main():
    node_path = "ns=2;s=TEST.PLC1"
    async with ClientASY(url=url) as clientAsy:
        Tags_List = await create_tags_dict(node_path, clientAsy)
        await asyncio.sleep(2)

        tasks = []
        for key, value in Tags_List.items():
            tasks.append(asyncio.create_task(get_value_from_dict(value, clientAsy)))
        RESULT = await asyncio.gather(*tasks)

        await asyncio.sleep(1)

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

问题分析 #

1. 冗余节点查找：get_value_from_dict中重复调用client.get_node(address)，但create_tags_dict已经将完整的Node对象存入字典，重复查找会额外消耗时间。
2. 任务创建开销：循环逐个调用asyncio.create_task创建上万任务，本身会带来显著的性能开销，导致tasks创建耗时过长。
3. 未利用批量读取特性：当前逐个读取标签的方式，没有用到OPC UA协议原生支持的批量读取接口，网络开销远大于批量操作。
4. 无并发限流：一次性发起上万请求可能导致服务器或客户端过载，反而拖慢整体速度。

优化方案 #

1. 复用已有的Node对象 #

直接使用create_tags_dict中存储的Node对象调用get_value，省去重复查找节点的步骤：

async def get_value_from_node(node):
    return await node.get_value()

2. 用列表推导式优化任务创建 #

列表推导式比循环append更高效，能减少任务创建的时间：

tasks = [asyncio.create_task(get_value_from_node(node)) for node in Tags_List.values()]

3. 使用Semaphore控制并发数 #

通过信号量限制同时执行的任务数量，避免过载：

sem = asyncio.Semaphore(100)  # 根据服务器性能调整并发数

async def get_value_with_sem(node):
    async with sem:
        return await node.get_value()

# 任务创建改为
tasks = [asyncio.create_task(get_value_with_sem(node)) for node in Tags_List.values()]

4. 核心优化：使用OPC UA批量读取接口 #

asyncua提供了read_values方法，可一次性读取多个节点的值，这是提升效率的关键：

async def main():
    node_path = "ns=2;s=TEST.PLC1"
    async with ClientASY(url=url) as clientAsy:
        node = clientAsy.get_node(node_path)
        variables = await node.get_variables()
        
        # 直接批量读取所有节点的值
        RESULT = await clientAsy.read_values(variables)
        
        await asyncio.sleep(1)

这种方式无需创建大量异步任务，通过一次批量请求获取所有值，能大幅降低网络和任务调度开销，完全可以满足1秒内读取50000个标签的需求。

最终优化后的完整代码 #

import asyncio
from asyncua import Client as ClientASY

url = "opc.tcp://127.0.0.1:49320"

async def main():
    node_path = "ns=2;s=TEST.PLC1"
    async with ClientASY(url=url) as clientAsy:
        # 获取目标节点下的所有变量节点
        parent_node = clientAsy.get_node(node_path)
        variables = await parent_node.get_variables()
        
        # 批量读取所有变量的值
        RESULT = await clientAsy.read_values(variables)
        
        # 可选：将结果与标签名对应
        tag_value_dict = {var.nodeid.Identifier: val for var, val in zip(variables, RESULT)}
        print(f"读取到{len(tag_value_dict)}个标签值")

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

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