23 KiB
23 KiB
cloud_interact 程序架构分析
分析范围:
process/cloud_apps/cloud_interact及其编译依赖的processers、reuse_utils
程序定位:储能/能源终端设备与云平台(或上级 EMS)之间的通信交互进程
作者:代码分析整理
日期:2026-06-10
1. 程序概述
cloud_interact 是云平台通信子系统的核心可执行程序,编译产物链接到 out/<TARGET_PRODUCT>/bin/cloud_interact。
1.1 核心职责
| 职责 | 说明 |
|---|---|
| 授权管理 | 启动时完成云平台/EMS 授权查询或授权接收,决定直连云还是透传 EMS |
| 云通信 | TCP 连接云平台,完成登录、心跳、上下行数据交互 |
| 数据上报 | 按优先级主动上报设备数据、告警、总览、统计等 |
| 下行控制 | 接收云平台下发的参数读写、控制指令、升级、SSH 等命令 |
| 中继转发 | EMS 终端将子终端(ABMS/PMS)数据转发至云平台 |
| 辅助任务 | 离线数据缓存上报、故障录波上报、远程 SSH 隧道管理 |
1.2 源码目录结构
process/cloud_apps/
├── cloud_interact/ # 本程序主体
│ ├── include/ # 通讯交互、授权、状态管理等头文件
│ └── src/ # 实现 + main.cpp
├── processers/ # 各帧类型数据处理器(策略模式)
│ ├── include/
│ └── src/
└── reuse_utils/ # 帧管理、配置、协议单例、下载等公共工具
├── include/
└── src/
1.3 外部动态库依赖
通过 CMakeLists.txt 链接的主要 SO 库:
cloud_protocol/cloud_communicator/commu_component— 协议编解码与 TCP 通信runtime_environ/running_data/cloud_data/cloud_command/cloud_repeater— 运行环境与共享内存数据upgrade_task/ssl_middleware/system_config— 升级、SSL、系统配置mutex_log/curl/ssh等基础设施库
2. 整体架构
2.1 线程模型
程序采用 单主线程 + 多工作线程 模型:
┌─────────────────────────────────────────────────────────────────┐
│ main 主线程 (200ms 轮询) │
│ - 信号处理 / 进程喂狗 / 授权状态监控 / 任务启停 / 日志刷新 │
└────────┬──────────────────┬──────────────────┬──────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ CCommuInteraction│ │ COfflineDataTask│ │ CFaultRecordTask│
│ (云通信核心线程) │ │ (离线数据线程) │ │ (故障录波线程) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
授权阶段额外线程(按终端类型):
- EMS: CAuthDistributeTask(授权分发给子终端,detached)
- 非 EMS: CAuthReceiveHandler 在 main 循环中轮询(监听 EMS 广播)
所有工作线程均继承 CBaseThread,实现 Handle() / FeedWatchDog() / SleepTime()。
2.2 分层架构
┌──────────────────────────────────────────────────────────┐
│ 应用层:main.cpp │
│ CAuthManager / CStateManager / CConfigManager │
├──────────────────────────────────────────────────────────┤
│ 任务层:CCommuInteraction / COfflineDataTask / ... │
├──────────────────────────────────────────────────────────┤
│ 通信协助层:CCommunicatorHelper / CForwardHelper │
│ CDownstreamCommand / CDataCache │
├──────────────────────────────────────────────────────────┤
│ 协议处理层:CFrameManager + CDataProcesser 子类群 │
├──────────────────────────────────────────────────────────┤
│ 基础设施层:CProtocolSinglton / SSLLibSinglton │
│ libcloud_* / librunning_data 等共享库 │
└──────────────────────────────────────────────────────────┘
3. 核心类与调用关系
3.1 类关系总览(Mermaid)
classDiagram
direction TB
class main {
+main()
}
class CStateManager {
+Instance()
+NotifyQuit()
+NeedQuit()
+NotifyInBlacklist()
+IsUnbindWithStation()
}
class CAuthManager {
+CreateInstance()
+Execute()
+ReceiveAuthBroadcast()
+Deinitialize()
}
class CCloudConnectivityDetector {
+Execute()
+IsTimeToExecute()
}
class CAuthQueryHandler {
+Execute()
+IsAuthPermitted()
}
class CAuthDistributeTask {
<<CBaseThread>>
+Handle()
}
class CAuthReceiveHandler {
+Execute()
}
class CConfigManager {
+Instance()
+Initialize()
+GetStationConfig()
}
class CProtocolSinglton {
+Instance()
+Create()
+Free()
}
class CCommuInteraction {
<<CBaseThread>>
+Initialize()
+Handle()
+ConnectCloud()
+LoginCloud()
+HandleDownstreamData()
+HandleUpstreamData()
}
class CFrameManager {
+CreateDataProcessers()
+DecodeDownstreamFrame()
+GetDataProcesserByFrameType()
+GetPriority2DataProcesser()
}
class CCommunicatorHelper {
+Connect()
+Send()
+Receive()
+Disconnect()
}
class CForwardHelper {
+ForwardDownstreamData()
+ForwardUpstreamData()
+ForwardFaultRecord()
+ForwardOfflineData()
}
class CDownstreamCommand {
+HandleDownstreamRequest()
+CheckDownstreamCommandResults()
}
class CDataCache {
+Instance()
+AddRequestData()
+PushOfflineDataCfg()
}
class CDataProcesser {
<<abstract>>
+EncodeUpstreamRequest()
+DecodeDownstreamRequest()
+TimeToEncodeUpstreamRequest()
}
class CLoginProcesser {
+CreateInstance()
}
class CHeartBeatProcesser {
+CreateInstance()
}
class COfflineDataTask {
<<CBaseThread>>
+Handle()
}
class CFaultRecordTask {
<<CBaseThread>>
+Handle()
}
class CRemoteSshHelper {
+Instance()
+Start()
+Stop()
+IsExpired()
}
main --> CStateManager
main --> CAuthManager
main --> CConfigManager
main --> CProtocolSinglton
main --> CCommuInteraction
main --> COfflineDataTask
main --> CFaultRecordTask
main --> CRemoteSshHelper
CAuthManager --> CCloudConnectivityDetector
CAuthManager --> CAuthQueryHandler
CAuthManager --> CAuthDistributeTask : EMS only
CAuthManager --> CAuthReceiveHandler : non-EMS
CCommuInteraction --> CFrameManager
CCommuInteraction --> CCommunicatorHelper
CCommuInteraction --> CForwardHelper
CCommuInteraction --> CDownstreamCommand
CCommuInteraction --> CDataCache
CCommuInteraction --> CStateManager
CCommuInteraction --> CConfigManager
CFrameManager --> CDataProcesser
CDataProcesser <|-- CLoginProcesser
CDataProcesser <|-- CHeartBeatProcesser
CDataProcesser <|-- CGeneralDataProcesser
CGeneralDataProcesser <|-- "60+ 具体 Processer"
CDownstreamCommand --> CFrameManager
CDownstreamCommand --> CCommunicatorHelper
CDownstreamCommand --> CForwardHelper
CDownstreamCommand --> CRemoteSshHelper
CForwardHelper --> CCommunicatorHelper
3.2 cloud_interact 模块类清单
| 类名 | 文件 | 类型 | 职责 |
|---|---|---|---|
CCommuInteraction |
CommuInteraction.h/cpp | 工作线程 | 云通信主循环:连接→登录→上下行收发 |
CCommuInterData |
CommuInteraction.cpp (内部) | 私有数据 | 持有 FrameManager、Communicator、Forward、DownstreamCommand |
CCommunicatorHelper |
CommunicatorHelper.h/cpp | 工具类 | 封装 ICloudCommunicator 的连接/发送/接收 |
CForwardHelper |
ForwardHelper.h/cpp | 工具类 | EMS 中继场景下子终端数据转发 |
CDownstreamCommand |
DownstreamCommand.h/cpp | 命令处理器 | 云平台下行控制指令分发与结果查询 |
CDataCache |
DataCahce.h/cpp | 单例 | 在线时长统计、请求追踪、离线/录波数据队列 |
CStateManager |
StateManager.h/cpp | 单例 | 进程退出标志、授权异常状态(解绑/拉黑/未注册等) |
CAuthManager |
AuthManager.h/cpp | 管理器 | 启动授权流程编排 |
CCloudConnectivityDetector |
CloudConnectivityDetector.h/cpp | 探测器 | 探测直连云或可连通服务器参数 |
CAuthQueryHandler |
AuthQueryTask.h/cpp | 授权查询 | 向云平台发送授权查询并写 StationConfig |
CAuthDistributeTask |
AuthDistributeTask.h/cpp | 工作线程(EMS) | 向子终端分发授权配置 |
CAuthReceiveHandler |
AuthReceiveTask.h/cpp | 监听器(非EMS) | 监听 EMS 授权广播(端口 10090) |
COfflineDataTask |
OfflineDataTask.h/cpp | 工作线程 | 断网期间离线数据本地缓存与补传 |
CFaultRecordTask |
FaultRecordTask.h/cpp | 工作线程 | 故障录波文件采集与上报 |
CRemoteSshHelper |
RemoteSshHelper.h/cpp | 单例 | 云平台远程 SSH 隧道启停与过期检测 |
CUpgradeTask |
UpgradeTask.h/cpp | 升级辅助 | 升级包下载与任务执行(被 CommuInteraction 调用) |
3.3 reuse_utils 模块类清单
| 类名 | 职责 |
|---|---|
CConfigManager |
加载 CloudParameters.xml、StationConfig,管理云通信参数 |
CFrameManager |
帧编解码路由,维护帧类型→Processor 映射与优先级队列 |
CFrameHelper |
通讯帧组包/拆包、授权配置读写等静态工具 |
CProtocolSinglton |
云协议 ICloudProtocol 单例生命周期管理 |
CSSLLibSinglton |
SSL 加解密库单例 |
CDowloadTask / CFtpDownloadTask / CSftpDowloadTask |
升级包下载 |
CUdpSender / CUdpReceiver |
UDP 通信(授权广播等场景) |
3.4 processers 模块 — 数据处理器体系
所有处理器继承自 CDataProcesser(抽象基类),通用数据类继承 CGeneralDataProcesser。
基类核心接口:
- 时机判断:
TimeToEncodeUpstreamRequest()/TimeToEncodeDownstreamResponse()等 - 编解码:
EncodeUpstreamRequest()/DecodeDownstreamRequest()等 - 数据填充:
FillUpstreamRequestData()/FillDownstreamRequestData()等
处理器按业务分组(头文件):
| 分组头文件 | 包含处理器 | 业务 |
|---|---|---|
LoginProcesser.h |
CLoginProcesser | 登录/登出 |
HeartBeatProcesser.h |
CHeartBeatProcesser | 心跳 |
DeviceManageProcessers.h |
CVersionInfoProcesser, CDeviceInfoProcesser, CDeviceStateProcesser | 版本/设备信息/状态 |
OverviewProcessers.h |
COverallOverviewProcesser, CTerminalOverviewProcesser, CESSProcesser 等 | 各类总览数据 |
DeviceDataProcessers.h |
CPCSProcesser, CPowerMeterProcesser, CSolarInverterProcesser 等 | 设备实时数据 |
BMSDataProcessers.h |
CBMSSummaryProcesser, CBMSExtremeProcesser, CBMSCellDetailsProcesser 等 | BMS 数据 |
AlarmEventProcessers.h |
CTeleSignalingProcesser, CRealtimeAlarmProcesser, CHistoryAlarmProcesser 等 | 告警/事件 |
EMSStrategyProcessers.h |
CDownstreamSchedulingTaskRead/WriteProcesser 等 | EMS 储能策略参数 |
EMSBatteryProcessers.h |
CDownstreamSafeProtectionRead/WriteProcesser 等 | 电池安全/校准参数 |
EMSControlProcessers.h |
CDownstreamEMSControlProcesser | EMS 控制指令 |
TariffParmsProcessers.h |
CDownstreamTariffTypeRead/WriteProcesser 等 | 电价参数 |
RemoteUpgradeProcessers.h |
CUpstreamUpgradeTaskQueryProcesser 等 | 远程升级 |
RemoteManageProcessers.h |
CDownstreamSshStart/StopProcesser | 远程 SSH |
OfflineProcessers.h |
CUpstreamOfflineRequestProcesser | 离线数据请求 |
FaultRecordProcessers.h |
CUpstreamFaultRecRequestProcesser | 故障录波 |
AuthorizationProcessers.h |
授权相关帧处理 | 授权交互 |
StatisticsProcessers.h |
CRevenueStatisticsProcesser 等 | 收益统计 |
SetupProcessers.h |
CDownstreamTermUpdateProcesser 等 | 终端更新/TOU 设置 |
注册机制: CFrameManager 构造函数中通过 mapType2ProcesserCreator 将 CLOUD_FRAME_TYPE_* 映射到各 Processor 的 CreateInstance 工厂函数,并按 mapFrameType2Priority 设定上报优先级。
4. 主流程工作流程
4.1 程序启动流程
flowchart TD
A[main 启动] --> B{运行环境已创建?}
B -- 否 --> Z1[退出 -100]
B -- 是 --> C[清除云平台授权状态]
C --> D[创建日志实例]
D --> E[创建必要目录]
E --> F{云功能已禁用?}
F -- 是 --> Z2[退出 0]
F -- 否 --> G[初始化运行时长统计]
G --> H{进程已运行?}
H -- 是 --> Z3[退出 0]
H -- 否 --> I[注册信号处理]
I --> J[CAuthManager::Execute 阻塞授权]
J -- 失败 --> Z4[退出 -5]
J -- 成功 --> K[重置 CStateManager 状态标志]
K --> L[CConfigManager::Initialize]
L --> M[CProtocolSinglton::Create]
M --> N[CCommuInteraction::Initialize]
N --> O[可选: OfflineDataTask / FaultRecordTask 初始化]
O --> P[启动 CThreadWatchDog]
P --> Q[进入主循环 while true]
4.2 授权管理流程
EMS 终端(直连云)
CCloudConnectivityDetector循环探测默认/XML 配置的服务器地址,直到 TCP 可连通CAuthQueryHandler循环发送授权查询,直到云平台返回许可- 写入
StationConfig.xml,通知librd_NotifyCloudAuthorized(1) - 启动
CAuthDistributeTask(detached 线程),向ems_cfg.xml配置的子终端分发授权 - 主流程继续初始化通信模块
非 EMS 终端(ABMS/PMS,直连或透传)
- 并行执行:
CAuthReceiveHandler:监听 EMS 授权广播(端口 10090)CCloudConnectivityDetector:探测云平台连通性
- 若先收到 EMS 广播 → 判定为透传终端,
librd_SetSelfTerminalForwarding(1) - 若先连通云平台 → 执行授权查询:
- 许可成功 → 直连终端
- 拒绝/无响应 → 继续等待 EMS 广播(透传)
- main 循环中持续调用
pAuthManager->ReceiveAuthBroadcast()接收授权更新
4.3 主循环职责(200ms 周期)
主线程在 while(true) 中执行:
- 检查
CStateManager::NeedQuit()— 信号触发退出 - 进程/线程喂狗
- 非 EMS:接收授权广播
- 授权异常恢复:若检测到解绑/拉黑/未注册/电站不匹配/协议未签署/断连需重授权:
- 停止
CCommuInteraction - 释放协议单例
- 重新执行
CAuthManager::Execute() - 重载配置、重建协议、重启通信
- 停止
- 启动
CCommuInteraction(若未运行) - 启动
COfflineDataTask/CFaultRecordTask(若启用且设备上报开启) - 检查守护进程状态、重启请求
- 检查
CRemoteSshHelper::IsExpired()并自动停止过期 SSH - 定时刷新日志、更新运行时长(5 分钟)
4.4 CCommuInteraction 通信线程流程
Handle() 每次循环(休眠 20ms)按固定顺序执行:
flowchart TD
H0[Handle 入口] --> H1{首次运行?}
H1 -- 是 --> H1a[LogoutCloud + 初始化 CDataCache]
H1 -- 否 --> H2
H1a --> H2{需要退出?}
H2 -- 是 --> EXIT[返回 -1]
H2 -- 否 --> H3[ConnectCloud TCP连接]
H3 -- 失败 --> RET0[返回 0 重试]
H3 -- 成功 --> H4[LoginCloud 登录]
H4 -- 失败 --> H4a{授权异常?}
H4a -- 是 --> EXIT
H4a -- 否 --> RET0
H4 -- 成功 --> H5[处理未完成升级任务]
H5 --> H6[CheckDownstreamCommandResults 查询控制结果]
H6 --> H7[HandleDownstreamData 接收并处理下行]
H7 --> H8[HandleUpstreamData 按优先级上报上行]
H8 --> H9{EMS转发有效?}
H9 -- 是 --> H10[ForwardUpstreamData 转发子终端上行]
H9 -- 否 --> H11
H10 --> H11[ForwardFaultRecord 转发故障录波]
H11 --> H12[ForwardOfflineData 转发离线数据]
H12 --> H13{5分钟无响应?}
H13 -- 是 --> H13a[LogoutCloud 重连]
H13 -- 否 --> H14[定时更新在线时长]
H14 --> OK[返回 1]
4.4.1 连接与登录
- ConnectCloud:通过
CCommunicatorHelper::Connect()建立 TCP,更新librd连接状态,指数退避重连 - LoginCloud:
CLoginProcesser构造登录请求帧CFrameHelper::EncodeUpstreamFrame组包CCommunicatorHelper::Send发送- 接收并解析登录响应,调用
CConfigManager::UpdateLoginResponse - 更新
CDataCache上线时间,设置bCloudLoggined = true
4.4.2 下行数据处理(云平台 → 终端)
ReceiveDownstreamFrame → CFrameManager::DecodeDownstreamFrame
→ 获取 CDataProcesser
→ 分支判断:
├─ 心跳响应 → DecodeDownstreamHeartBeatResponse
├─ 离线数据响应 → DecodeDownstreamOfflineDataResponse
├─ 故障录波响应 → DecodeDownstreamFaultRecordResponse
├─ 升级任务查询响应 → DecodeDownstreamUpgradeTaskQueryResponse
├─ 下行命令 → CDownstreamCommand::HandleDownstreamRequest
│ ├─ 终端更新
│ ├─ 升级任务通知
│ ├─ 参数读写(策略/电池/电价/SSH等)
│ └─ 控制指令
└─ 需转发 → CForwardHelper::ForwardDownstreamData(EMS→子终端)
4.4.3 上行数据处理(终端 → 云平台)
遍历 CFrameManager::GetPriority2DataProcesser() 优先级 map(数值越小优先级越高):
- 调用
pProcesser->TimeToEncodeUpstreamRequest()判断是否到时上报 - 填充数据
FillUpstreamRequestData() - 编码并组帧
EncodeUpstreamRequest+CFrameHelper::EncodeUpstreamFrame SendUpstreamFrame发送- 等待响应帧,解析并更新上报状态
- 特殊帧:心跳、升级任务查询、离线/录波响应等单独处理
4.5 授权异常自愈机制
当云平台在登录或交互中返回以下状态时,CStateManager 置位,通信线程退出,主循环触发重授权:
| 状态标志 | 触发场景 |
|---|---|
IsAgreementUnsigned |
协议未签署(V2400+ 协议) |
IsNotRegisterd |
终端未在云平台登记 |
IsUnbindWithStation |
与电站解绑 |
IsInBlacklist |
进入黑名单 |
IsStationDismatched |
电站 ID 不匹配 |
IsDisconnectedForReobtainAuth |
连接断开需重新获取授权 |
5. 关键协作时序
5.1 正常通信时序
sequenceDiagram
participant Main as main线程
participant CI as CCommuInteraction
participant CH as CCommunicatorHelper
participant FM as CFrameManager
participant DP as CDataProcesser
participant Cloud as 云平台
Main->>CI: Start(true)
loop 每20ms
CI->>CH: Connect()
CH->>Cloud: TCP连接
CI->>FM: GetDataProcesser(LOGIN)
FM->>DP: CLoginProcesser
DP->>DP: EncodeUpstreamRequest
CI->>CH: Send(登录帧)
CH->>Cloud: 登录请求
Cloud-->>CH: 登录响应
CI->>FM: DecodeDownstreamFrame
CI->>FM: 遍历优先级队列
FM->>DP: TimeToEncodeUpstreamRequest?
DP->>DP: Fill + Encode
CI->>CH: Send(数据帧)
CH->>Cloud: 上行数据
Cloud-->>CH: 下行命令/响应
CI->>FM: DecodeDownstreamFrame
end
5.2 下行控制指令时序
sequenceDiagram
participant Cloud as 云平台
participant CI as CCommuInteraction
participant DC as CDownstreamCommand
participant DP as CDataProcesser
participant RD as librunning_data
Cloud->>CI: 下行控制/参数帧
CI->>DC: HandleDownstreamRequest
DC->>DP: DecodeDownstreamRequest
DP->>RD: 写入控制/参数到运行数据
DC->>DP: EncodeUpstreamResponse
CI->>Cloud: 上行响应(立即确认)
Note over DC,RD: 异步执行控制指令
loop 周期检查
DC->>DC: CheckDownstreamCommandResults
DC->>RD: 查询执行结果
DC->>Cloud: 上报最终结果
end
6. 配置与编译开关
| 宏/开关 | 位置 | 作用 |
|---|---|---|
__ENABLE_CLOUD_TASK__ |
main.cpp | 启用 CCommuInteraction |
__ENABLE_AUTH_TASK__ |
main.cpp | 启用授权管理 |
__ENABLE_OFFLINE_TASK__ |
OfflineDataTask.h | 启用离线数据任务 |
__ENABLE_FAULT_TASK__ |
FaultRecordTask.h | 启用故障录波任务 |
CURRENT_TERMINAL_TYPE |
CMake 产品宏 | 区分 EMS / PMS / ABMS 编译 |
CURRENT_CLOUD_PROTOCOL_VERSION |
协议版本宏 | 控制协议特性(如协议签署) |
IS_DEVICE_UPLOAD_ENABLE |
配置宏 | 控制离线/录波任务启动 |
7. 设计特点总结
- 策略模式 + 工厂注册:新增云平台帧类型只需在
processers添加 Processor 并在CFrameManager注册,无需修改CCommuInteraction主逻辑 - 下行命令集中处理:
CDownstreamCommand封装所有下行控制,与上报逻辑解耦 - 优先级队列上报:
mapPriority2DataProcesser保证登录、告警等高优先级数据优先发送 - 双模连接:通过授权流程自动识别直连云 / EMS 透传两种拓扑
- 授权自愈:主循环监控授权状态,异常时自动停服→重授权→重启通信
- Pimpl 惯用法:各类均使用
*Private/*Data内部类隐藏实现细节
8. 相关文档
- 类关系详图:见 cloud_interact_class_reference.md
- 云平台协议帧定义:参考
include/interface/ICloudProtocol.h及协议文档 - EMS 子终端配置:
ems_cfg.xml(授权分发) - 云通信参数:
CloudParameters.xml/StationConfig.xml