OPCUA与asyncua实战入门：从零构建一个异步数据服务器与客户端

1. 为什么选择OPC UA和asyncua？

在工业物联网和自动化控制领域，设备间的数据通信一直是个头疼的问题。不同厂商的设备使用不同的协议，就像一群人说着不同的方言，沟通起来特别费劲。OPC UA（Open Platform Communications Unified Architecture）就是为了解决这个问题而生的，它就像工业领域的"普通话"，让各种设备能够用同一种语言交流。

而asyncua则是OPC UA的Python实现，它有几个特别吸引人的特点：

1. 纯Python实现：不需要复杂的编译环境，安装即用
2. 异步IO支持：基于asyncio，适合高并发的工业场景
3. 简单易用：API设计直观，学习曲线平缓
4. 开源免费：社区活跃，问题容易得到解决

我去年在一个智能工厂项目中就用到了asyncua，当时需要在30多台设备间建立实时数据交换。传统方案要么太贵，要么开发周期长，而asyncua让我们在两周内就搭建起了原型系统，实测下来非常稳定。

2. 环境准备与安装

2.1 Python环境要求

asyncua需要Python 3.7及以上版本。我强烈建议使用虚拟环境来管理项目依赖，这样可以避免包冲突问题。下面是创建虚拟环境的命令：

# 创建虚拟环境 python -m venv opcua_env # 激活虚拟环境 # Windows opcua_env\Scripts\activate # Linux/MacOS source opcua_env/bin/activate

2.2 安装asyncua

安装asyncua非常简单，一条pip命令就能搞定：

pip install asyncua

如果你需要用到额外的功能，比如历史数据记录，可以安装可选依赖：

pip install asyncua[history]

我在Windows和Linux上都测试过安装过程，基本不会遇到什么问题。唯一需要注意的是，在某些企业内网环境下，可能需要先配置好pip的代理设置。

3. 构建OPC UA服务器

3.1 基础服务器搭建

让我们从一个最简单的服务器开始。新建一个文件server.py，写入以下代码：

import logging import asyncio from asyncua import Server logging.basicConfig(level=logging.INFO) logger = logging.getLogger('asyncua') async def main(): # 创建服务器实例 server = Server() await server.init() # 设置服务器端点地址 server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/") # 设置服务器名称 server.set_server_name("My First OPC UA Server") # 注册命名空间 uri = "http://mycustom.namespace.com" idx = await server.register_namespace(uri) logger.info("Server started at opc.tcp://0.0.0.0:4840/freeopcua/server/") async with server: while True: await asyncio.sleep(1) if __name__ == "__main__": asyncio.run(main())

这段代码做了以下几件事：

1. 创建了一个OPC UA服务器实例
2. 设置监听地址为0.0.0.0:4840
3. 注册了一个自定义命名空间
4. 启动服务器并保持运行

你可以通过运行python server.py来启动服务器。启动后，用专业的OPC UA客户端（如UaExpert）就能连接到这个服务器了。

3.2 添加变量和对象

光有服务器还不够，我们需要添加一些实际的数据点。修改main函数，添加以下内容：

async def main(): # ...之前的代码不变... # 获取Objects节点 objects = server.nodes.objects # 添加一个自定义对象 my_obj = await objects.add_object(idx, "MyDevice") # 添加一些变量 temperature = await my_obj.add_variable(idx, "Temperature", 25.0) pressure = await my_obj.add_variable(idx, "Pressure", 1.0) status = await my_obj.add_variable(idx, "Status", "Idle") # 设置变量为可写 await temperature.set_writable() await pressure.set_writable() await status.set_writable() # ...之后的代码不变...

现在我们的服务器就有了三个变量：温度、压力和状态。这些变量都可以通过客户端读取和写入。在实际项目中，你可能会把这些变量绑定到实际的设备数据上。

4. 构建OPC UA客户端

4.1 基础客户端实现

有了服务器，我们还需要一个客户端来测试通信。新建一个文件client.py，写入以下代码：

import asyncio import logging from asyncua import Client logging.basicConfig(level=logging.INFO) logger = logging.getLogger('asyncua') async def main(): client = Client(url="opc.tcp://localhost:4840/freeopcua/server/") try: async with client: # 获取命名空间索引 uri = "http://mycustom.namespace.com" idx = await client.get_namespace_index(uri) # 读取变量 temp_node = await client.nodes.root.get_child([ "0:Objects", f"{idx}:MyDevice", f"{idx}:Temperature" ]) pressure_node = await client.nodes.root.get_child([ "0:Objects", f"{idx}:MyDevice", f"{idx}:Pressure" ]) current_temp = await temp_node.read_value() current_pressure = await pressure_node.read_value() logger.info(f"Current Temperature: {current_temp}") logger.info(f"Current Pressure: {current_pressure}") except Exception as e: logger.error(f"Connection failed: {e}") if __name__ == "__main__": asyncio.run(main())

这个客户端会连接到我们之前创建的服务器，并读取温度和压力变量的值。运行它之前，确保服务器已经在运行。

4.2 写入变量和订阅变化

除了读取变量，客户端还可以修改变量值和订阅变量变化。在main函数中添加以下代码：

async def main(): client = Client(url="opc.tcp://localhost:4840/freeopcua/server/") try: async with client: # ...之前的代码不变... # 写入新值 new_temp = current_temp + 1.0 await temp_node.write_value(new_temp) logger.info(f"Temperature updated to: {new_temp}") # 订阅变量变化 subscription = await client.create_subscription(500, handler) handle = await subscription.subscribe_data_change(temp_node) # 保持连接以接收通知 await asyncio.sleep(10000) except Exception as e: logger.error(f"Error: {e}") # 变化处理函数 def handler(node, val, data): print(f"Data change on {node}: {val}")

现在，每当温度值发生变化时，handler函数就会被调用，打印出新的值。这在实际应用中非常有用，比如当某个设备参数超出安全范围时，可以立即触发警报。

5. 实际应用中的进阶技巧

5.1 安全配置

在生产环境中，安全性至关重要。OPC UA支持多种安全策略，我们可以这样配置：

async def main(): server = Server() # 启用基本安全策略 await server.init() server.set_security_policy([ ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt, ua.SecurityPolicyType.Basic256Sha256_Sign ]) # 加载证书和私钥 await server.load_certificate("server_cert.pem") await server.load_private_key("server_key.pem") # ...其他代码...

客户端连接时也需要提供相应的证书：

client = Client(url="opc.tcp://localhost:4840/") client.set_security_string("Basic256Sha256,SignAndEncrypt,cert.pem,key.pem")

5.2 性能优化

当需要处理大量数据时，性能优化就很重要了。这里有几个我实践过的技巧：

1. 批量读取：使用read_values()代替多次read_value()
2. 合理设置订阅参数：采样间隔不宜过短
3. 使用历史数据：对于不常变化的数据，可以启用历史记录功能

# 批量读取示例 nodes_to_read = [temp_node, pressure_node, status_node] values = await client.read_values(nodes_to_read)

5.3 错误处理与重连机制

网络不稳定是常见问题，实现自动重连很有必要：

async def run_client(): while True: try: await main() except ConnectionError: logger.warning("Connection lost, retrying in 5 seconds...") await asyncio.sleep(5) except Exception as e: logger.error(f"Unexpected error: {e}") break if __name__ == "__main__": asyncio.run(run_client())

6. 调试与问题排查

6.1 常见问题解决

在实际开发中，你可能会遇到这些问题：

1. 连接失败：检查防火墙设置和端口是否开放
2. 证书错误：确保证书有效且时间正确
3. 权限问题：检查用户是否有读写权限

我遇到过最棘手的问题是证书时间不同步，导致安全连接失败。后来发现是服务器时间比实际时间慢了5分钟，调整后问题解决。

6.2 日志配置

合理的日志配置能大大简化调试过程：

logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('opcua.log'), logging.StreamHandler() ] )

这样既能将日志输出到控制台，又能保存到文件中，方便后续分析。

6.3 使用Wireshark抓包分析

对于复杂的通信问题，可以使用Wireshark抓取OPC UA协议包：

1. 安装Wireshark
2. 过滤条件设置为opcua
3. 分析通信过程中的具体报文

这个方法帮我解决过多次协议层面的疑难杂症，特别是在安全策略配置出错时特别有用。