Files
he/process/doc/cloud_interact_analysis.md
T

579 lines
23 KiB
Markdown
Raw Normal View History

2026-06-22 15:35:18 +08:00
# 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)
```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 程序启动流程
```mermaid
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 终端(直连云)
1. `CCloudConnectivityDetector` 循环探测默认/XML 配置的服务器地址,直到 TCP 可连通
2. `CAuthQueryHandler` 循环发送授权查询,直到云平台返回许可
3. 写入 `StationConfig.xml`,通知 `librd_NotifyCloudAuthorized(1)`
4. 启动 `CAuthDistributeTask`(detached 线程),向 `ems_cfg.xml` 配置的子终端分发授权
5. 主流程继续初始化通信模块
#### 非 EMS 终端(ABMS/PMS,直连或透传)
1. 并行执行:
- `CAuthReceiveHandler`:监听 EMS 授权广播(端口 10090)
- `CCloudConnectivityDetector`:探测云平台连通性
2. **若先收到 EMS 广播** → 判定为透传终端,`librd_SetSelfTerminalForwarding(1)`
3. **若先连通云平台** → 执行授权查询:
- 许可成功 → 直连终端
- 拒绝/无响应 → 继续等待 EMS 广播(透传)
4. main 循环中持续调用 `pAuthManager->ReceiveAuthBroadcast()` 接收授权更新
### 4.3 主循环职责(200ms 周期)
主线程在 `while(true)` 中执行:
1. 检查 `CStateManager::NeedQuit()` — 信号触发退出
2. 进程/线程喂狗
3. 非 EMS:接收授权广播
4. **授权异常恢复**:若检测到解绑/拉黑/未注册/电站不匹配/协议未签署/断连需重授权:
- 停止 `CCommuInteraction`
- 释放协议单例
- 重新执行 `CAuthManager::Execute()`
- 重载配置、重建协议、重启通信
5. 启动 `CCommuInteraction`(若未运行)
6. 启动 `COfflineDataTask` / `CFaultRecordTask`(若启用且设备上报开启)
7. 检查守护进程状态、重启请求
8. 检查 `CRemoteSshHelper::IsExpired()` 并自动停止过期 SSH
9. 定时刷新日志、更新运行时长(5 分钟)
### 4.4 CCommuInteraction 通信线程流程
`Handle()` 每次循环(休眠 20ms)按固定顺序执行:
```mermaid
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**:
1. `CLoginProcesser` 构造登录请求帧
2. `CFrameHelper::EncodeUpstreamFrame` 组包
3. `CCommunicatorHelper::Send` 发送
4. 接收并解析登录响应,调用 `CConfigManager::UpdateLoginResponse`
5. 更新 `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(数值越小优先级越高):
1. 调用 `pProcesser->TimeToEncodeUpstreamRequest()` 判断是否到时上报
2. 填充数据 `FillUpstreamRequestData()`
3. 编码并组帧 `EncodeUpstreamRequest` + `CFrameHelper::EncodeUpstreamFrame`
4. `SendUpstreamFrame` 发送
5. 等待响应帧,解析并更新上报状态
6. 特殊帧:心跳、升级任务查询、离线/录波响应等单独处理
### 4.5 授权异常自愈机制
当云平台在登录或交互中返回以下状态时,`CStateManager` 置位,通信线程退出,主循环触发重授权:
| 状态标志 | 触发场景 |
|---------|---------|
| `IsAgreementUnsigned` | 协议未签署(V2400+ 协议) |
| `IsNotRegisterd` | 终端未在云平台登记 |
| `IsUnbindWithStation` | 与电站解绑 |
| `IsInBlacklist` | 进入黑名单 |
| `IsStationDismatched` | 电站 ID 不匹配 |
| `IsDisconnectedForReobtainAuth` | 连接断开需重新获取授权 |
---
## 5. 关键协作时序
### 5.1 正常通信时序
```mermaid
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 下行控制指令时序
```mermaid
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. 设计特点总结
1. **策略模式 + 工厂注册**:新增云平台帧类型只需在 `processers` 添加 Processor 并在 `CFrameManager` 注册,无需修改 `CCommuInteraction` 主逻辑
2. **下行命令集中处理**:`CDownstreamCommand` 封装所有下行控制,与上报逻辑解耦
3. **优先级队列上报**:`mapPriority2DataProcesser` 保证登录、告警等高优先级数据优先发送
4. **双模连接**:通过授权流程自动识别直连云 / EMS 透传两种拓扑
5. **授权自愈**:主循环监控授权状态,异常时自动停服→重授权→重启通信
6. **Pimpl 惯用法**:各类均使用 `*Private` / `*Data` 内部类隐藏实现细节
---
## 8. 相关文档
- 类关系详图:见 [cloud_interact_class_reference.md](./cloud_interact_class_reference.md)
- 云平台协议帧定义:参考 `include/interface/ICloudProtocol.h` 及协议文档
- EMS 子终端配置:`ems_cfg.xml`(授权分发)
- 云通信参数:`CloudParameters.xml` / `StationConfig.xml`