admin/he
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
he/process/doc/cloud_interact_analysis.md
T

579 lines
23 KiB
Markdown
Raw Permalink 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`
Reference in New Issue Copy Permalink