为什么需要OpcUA订阅通信？
在工业物联网（IIoT）场景中，OPC UA（Open Platform Communications Unified Architecture）已成为工业设备通信的事实标准。它解决了传统OPC（OLE for Process Control）的跨平台、安全性和扩展性问题。
核心需求：设备数据实时监控 → 需要高效的数据订阅机制，而非轮询（Polling），避免网络和服务器负载过高。
OpcListener类实现了以下核心功能：
- 自动连接/重连OPC UA服务器
- 订阅管理（创建/删除订阅）
- 节点监视项管理（添加/删除监视项）
- 数据变化回调处理
- 断线自动重连机制
架构设计
核心成员变量
UA_Client*m_client;// OPC UA客户端实例std::string m_endpointUrl;// 服务器端点URLboolm_running=false;// 服务运行状态boolm_connected=false;// 连接状态UA_UInt32 m_subscriptionId=0;// 当前订阅ID// 重连配置intm_reconnectIntervalMs=1000;intm_maxReconnectAttempts=ContinueRetry;intm_currentReconnectAttempts=0;// 监视项存储std::unordered_map<UA_UInt32,OpcListenerItem>m_listenerItems;std::mutex m_itemMutex;// 保护监视项的互斥锁关键数据结构
structOpcListenerItem{UA_Variant value;// 当前值UA_NodeId nodeId;// 节点IDOpcListenerHandler handler;// 回调函数UA_UInt32 monitoredItemId;// 监视项IDstd::string nodeName;// 节点名称};核心功能实现解析
客户端初始化与连接
OpcListener::OpcListener(){m_client=UA_Client_new();UA_ClientConfig*config=UA_Client_getConfig(m_client);config->timeout=DefaultTimeout;UA_ClientConfig_setDefault(config);// 设置默认配置}连接与重连机制
核心逻辑：状态机驱动
关键代码解析：
boolOpcListener::internalConnect(){// 1. 连接服务器UA_StatusCode retval=UA_Client_connect(m_client,(char*)m_endpointUrl.c_str());// 2. 创建订阅（核心！）UA_CreateSubscriptionRequest request=UA_CreateSubscriptionRequest_default();request.requestedPublishingInterval=m_intervalMs;// 采样间隔request.requestedLifetimeCount=1000;// 订阅生命周期（1000次心跳）// ...其他参数UA_CreateSubscriptionResponse response=UA_Client_Subscriptions_create(m_client,request,...);m_subscriptionId=response.subscriptionId;// 3. 重新创建之前订阅的监视项for(auto&item:m_listenerItems){UA_MonitoredItemCreateResult result=UA_Client_MonitoredItems_createDataChange(...);// 重置监控项IDitem.second.monitoredItemId=result.monitoredItemId;}}- 订阅（Subscription）：OPC UA服务器端的核心机制，客户端通过订阅获取数据变更（而非轮询）。
- requestedPublishingInterval：设置数据发布间隔（毫秒），与m_intervalMs一致。
- requestedLifetimeCount：订阅的“心跳”次数（服务器每1000次心跳检查一次订阅存活）。
- 自动恢复：internalConnect() 会重新创建所有监视项，确保重连后监听不丢失。
监听管理器
addListener添加监听器流程：
boolOpcListener::addListener(constUA_NodeId&nodeId,OpcListenerHandler handler){// 1. 构建监视项请求UA_MonitoredItemCreateRequest itemRequest=UA_MonitoredItemCreateRequest_default(nodeId);itemRequest.requestedParameters.samplingInterval=m_intervalMs;itemRequest.requestedParameters.discardOldest=true;// 丢弃旧数据// 2. 创建监视项（重试机制）for(inti=0;i<MaxRetryCount;++i){UA_MonitoredItemCreateResult result=UA_Client_MonitoredItems_createDataChange(m_client,m_subscriptionId,UA_TIMESTAMPSTORETURN_BOTH,itemRequest,this,DataChangeCallback,nullptr);if(result.statusCode==UA_STATUSCODE_GOOD){// 3. 保存监视项信息OpcListenerItem item={variant,nodeId,handler,result.monitoredItemId,...};m_listenerItems.insert({result.monitoredItemId,item});returntrue;}}}- discardOldest=true：当队列满时丢弃旧数据，避免内存溢出（工业场景常见）。
- 重试机制：最多MaxRetryCount次重试，避免瞬时网络抖动导致失败。
- 线程安全：m_itemMutex 保护m_listenerItems，防止多线程并发操作。
数据回调处理
DataChangeCallback+ 线程池 回调函数核心逻辑：
voidOpcListener::DataChangeCallback(...){// 1. 从m_listenerItems获取handlerstd::lock_guardlock(instance->m_itemMutex);handler=instance->m_listenerItems[monId].handler;// 2. 复制数据（避免直接操作OPC UA内部数据）UA_Variant_copy(&value->value,&instance->m_listenerItems[monId].value);// 3. 提交到线程池执行（避免阻塞OPC UA事件循环）gT.push([=](){handler(nodeId,value);// 用户自定义处理});}为什么需要线程池？
- 如果用户处理耗时（如写数据库、AI分析），会阻塞OPC UA网络事件处理。
断开连接与资源清理
disconnect断开连接并清理资源
voidOpcListener::disconnect(){m_connected=false;if(m_subscriptionId!=0){UA_Client_Subscriptions_deleteSingle(m_client,m_subscriptionId);// 删除订阅m_subscriptionId=0;}UA_Client_disconnect(m_client);// 断开连接// 清理监听项for(auto&item:m_listenerItems){UA_Variant_clear(&item.second.value);UA_NodeId_clear(&item.second.nodeId);}m_listenerItems.clear();}- 顺序清理：先删除订阅 → 再断开连接 → 最后清理监听项。
订阅通信的核心优势
| 传统轮询方式 | 订阅 | 价值 |
|---|---|---|
| 服务器负载高（每秒请求N次） | 服务器负载低 | （仅变更时推送） 降低服务器CPU |
| 数据延迟高 | 数据延迟低 | 满足实时控制需求 |
| 网络带宽浪费（大量重复请求） | 带宽利用率高（仅发送变更数据） | 减少网络流量 |
OPC UA订阅原理：
- 客户端创建订阅 → 服务器分配订阅ID
- 客户端添加监视项 → 服务器注册节点监控
- 数据变更时 → 服务器通过订阅ID推送数据
- 客户端处理数据 → 用户逻辑执行
使用示例：
#include"src/OpcListener.hpp"classTestHandler{public:voidhandler(constUA_NodeId&nodeId,constUA_Variant&value){std::cout<<std::string((char*)nodeId.identifier.string.data,nodeId.identifier.string.length)<<" value changed."<<std::endl;if(value.type==&UA_TYPES[UA_TYPES_INT16]){std::cout<<"New value: "<<*(UA_Int16*)value.data<<std::endl;}else{std::cout<<"Error: Unknown type."<<std::endl;}}};intmain(){OpcListener listener;listener.setIntervalMs(1000);listener.setReconnectParameters(1000,5);listener.connect("opc.tcp://localhost:4840");TestHandler testHandler;listener.addListener(4,"test",&testHandler,&TestHandler::handler);while(listener.isRunning()){std::this_thread::sleep_for(std::chrono::seconds(1));}return0;}源码地址：https://gitcode.com/IT_Grey_Cat/OpcListener
