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
admin 74e4f32cdf 提交
2026-06-22 15:35:18 +08:00

23 KiB
Raw Permalink Blame History

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 终端（直连云）

  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）按固定顺序执行：

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 正常通信时序

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. 设计特点总结

  1. 策略模式 + 工厂注册：新增云平台帧类型只需在 processers 添加 Processor 并在 CFrameManager 注册，无需修改 CCommuInteraction 主逻辑
  2. 下行命令集中处理：CDownstreamCommand 封装所有下行控制，与上报逻辑解耦
  3. 优先级队列上报：mapPriority2DataProcesser 保证登录、告警等高优先级数据优先发送
  4. 双模连接：通过授权流程自动识别直连云 / EMS 透传两种拓扑
  5. 授权自愈：主循环监控授权状态，异常时自动停服→重授权→重启通信
  6. Pimpl 惯用法：各类均使用 *Private / *Data 内部类隐藏实现细节

8. 相关文档

  • 类关系详图：见 cloud_interact_class_reference.md
  • 云平台协议帧定义：参考 include/interface/ICloudProtocol.h 及协议文档
  • EMS 子终端配置：ems_cfg.xml（授权分发）
  • 云通信参数：CloudParameters.xml / StationConfig.xml
Reference in New Issue View Git Blame Copy Permalink