1084 lines
60 KiB
Markdown
1084 lines
60 KiB
Markdown
# SoftBus OPC 软件总体架构
|
||
|
||
> 本文档描述 SoftBus OPC 软总线系统的术语定义、软件背景、总体范围、架构概览及 Daemon 核心模块设计。
|
||
> 工程路径：`softbus_OPC/`。相关补充文档见 [docs/README.md](README.md)。
|
||
|
||
---
|
||
|
||
## 第 0 章 术语、缩写与软件背景
|
||
|
||
### 0.1 术语与缩写词
|
||
|
||
本章汇总全文涉及的专业术语与缩写，按类别编排。英文名称与代码标识保持一致。
|
||
|
||
#### 0.1.1 系统与进程
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| SoftBus OPC | — | 本软总线工业数据总线 Phase 1 原型工程名 |
|
||
| 软总线 | Soft Bus | 抽象异构现场总线，提供统一数据语义与传输编排的中间件 |
|
||
| Daemon | softbus_daemon | 中心守护进程：上行接入、协议解析、OPC UA 暴露、命令分发、在线监测 |
|
||
| Edge | softbus_edge | 边缘代理进程，部署于现场侧，负责物理设备采集与写回 |
|
||
| CoreService | — | Daemon 唯一编排入口，管理各子系统生命周期与主循环 |
|
||
| EdgeRuntime | — | Edge 运行时核心类，集成采集、上行、下行与配置热加载 |
|
||
| Phase 1 / 2 / 3 / 4 | — | 能力交付阶段：采集透传 → 订阅/管理 → 融合 → 恢复 |
|
||
| 南向 | Southbound | 面向现场设备与传感器的数据接入方向 |
|
||
| 北向 | Northbound | 面向 OPC UA、MES/SCADA 等上层应用的数据开放方向 |
|
||
|
||
#### 0.1.2 域模型与配置
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| 物理域 | Physical Domain | 通道、设备、寄存器、COB-ID 等贴近硬件的配置与数据标识；由 edge 加载 |
|
||
| 逻辑域 | Logical Domain | OPC UA 地址空间、逻辑点、对象模型五层路径；由 daemon 加载 |
|
||
| 双域 | Dual Domain | 物理配置与逻辑模型分离，经 Binding 映射关联 |
|
||
| 物理通道 | Physical Channel | `physical_channels.json` 中定义的串口/TCP/CAN 接口实例 |
|
||
| 物理设备 | Physical Device | 挂载于通道上的从站或 CANopen 节点等设备实例 |
|
||
| 物理点 | Physical Point | 最小采集单元，标识为 `physicalPointId`（运行时带 `edgeId` 前缀） |
|
||
| 逻辑点 | Logical Point | 北向语义测点，标识为 `logicalPointId`，对应 OPC UA Variable |
|
||
| Binding / 绑定 | BindingRegistry | 物理点 ID、逻辑点 ID 与 edgeId 的映射关系 |
|
||
| 全局绑定 | bindings_global.json | daemon 侧绑定表，决定上行映射与下行路由 |
|
||
| 本地绑定 | bindings_local.json | edge 侧可选绑定，用于下行写命令备用解析 |
|
||
| 逻辑点表 | logical_points.json | daemon 逻辑点与 OPC UA 节点语义定义 |
|
||
| 对象模型 | ObjectModelTree | 五层树形结构：Site / System / Asset / Device / Point |
|
||
| ObjectRef | — | 五层寻址路径字符串，如 `DemoSite/DemoSystem/DemoAsset/EdgeDevice_01/MotorTemperature` |
|
||
| edgeId | — | 边缘实例唯一标识，用于 uplink 注册与 daemon 路由 |
|
||
| deviceId | — | 逻辑设备名，用于主题格式化与 ACK 身份 |
|
||
| sourceId | — | RawPacket 中的数据源标识，通常为 `{edgeId}/{physicalPointId}` |
|
||
| topic_rule | topic_rule.json | Bus 主题模板：data / command / ack / event / heartbeat |
|
||
|
||
#### 0.1.3 数据模型与消息
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| RawPacket | — | 边缘上行边界类型：traceId、timestampNs、runtimeDeviceId、sourceId、protocolTag、payload |
|
||
| DOMMessage | Document Object Model Message | 协议插件解析后的统一语义消息 |
|
||
| SoftbusDataPoint | — | Bus 上发布的标准化测点：deviceId、pointId、value、unit、objectRef、sourceTimestampNs |
|
||
| ExecuteCommand | — | 下行控制命令：objectRef、command、params、requestId、traceId |
|
||
| SoftbusAck | — | edge 执行写命令后的应答：success、error、executedValue 等 |
|
||
| WireMessage | — | TCP 传输层消息封装，类型含 RawPacket、ExecuteCommand、heartbeat 等 |
|
||
| traceId | — | 贯穿上行/下行链路的追踪标识，便于日志关联 |
|
||
| protocolTag | — | RawPacket 中的协议标签，如 `modbus`、`canopen` |
|
||
| payload | — | RawPacket 中的原始协议字节载荷 |
|
||
| quality / 质量码 | Good / Uncertain / Bad | 数据可信度标识（Phase 2 规划映射至 OPC StatusCode） |
|
||
| deadband / 死区 | — | edge 变化过滤阈值，工程值变化小于死区时不重复上报 |
|
||
| reportOnChange | — | 仅变化时上报，配合 ChangeFilter 减少上行流量 |
|
||
| scale / 缩放系数 | — | 原始寄存器值与工程值之间的换算系数 |
|
||
| unbound 主题 | edge/unbound/raw/... | 未在 bindings_global 中绑定的物理点 RawPacket 发布主题 |
|
||
|
||
#### 0.1.4 传输与通信
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| UplinkWire | — | edge 与 Daemon 之间的 TCP JSON 行传输协议 |
|
||
| Uplink / 上行 | — | edge → daemon 方向的数据与控制应答 |
|
||
| 下行 | Downlink | daemon → edge 方向的 ExecuteCommand 写回 |
|
||
| TcpUplinkClient | — | edge 侧 TCP 上行客户端（含独立 ioThread） |
|
||
| TcpUplinkServer | — | daemon 侧 TCP 上行监听服务，默认端口 9000 |
|
||
| TransportBusBridge | — | TCP 线消息与 Bus 主题之间的双向桥接 |
|
||
| UplinkClient | — | edge 对 TcpUplinkClient 的封装 |
|
||
| Bus | softbus_bus | daemon 进程内 Pub/Sub 消息总线 |
|
||
| Pub/Sub | Publish / Subscribe | 发布/订阅模式，按主题模式过滤消息 |
|
||
| 心跳 | Heartbeat | edge 周期性发送的在线保活消息 |
|
||
| heartbeatPeriodMs | — | edge 心跳发送周期，默认 1000 ms |
|
||
| JSONL | JSON Lines | 每行一条 JSON 的日志/时序落盘格式 |
|
||
| TLS / mTLS | — | 传输层安全（Phase 2+ 规划） |
|
||
|
||
#### 0.1.5 协议与现场总线
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| Modbus RTU | — | 串口 Modbus，经 libmodbus 或 RTU-over-TCP 接入 |
|
||
| Modbus TCP | — | 以太网 Modbus 主从协议 |
|
||
| RTU-over-TCP | tcp_rtu / rtu_tcp | 在 TCP 连接上传输 Modbus RTU 帧（含 CRC） |
|
||
| CAN 2.0A/B | — | 控制器局域网标准帧格式 |
|
||
| CANopen | — | 基于 CAN 的工业协议，含 NMT、PDO、SDO |
|
||
| SocketCAN | — | Linux 内核 CAN 套接字接口（如 emuccan0） |
|
||
| PDO | Process Data Object | CANopen 过程数据对象，周期性广播 |
|
||
| SDO | Service Data Object | CANopen 服务数据对象，点对点读写对象字典 |
|
||
| COB-ID | CAN Object Identifier | CAN 帧标识符，PDO 订阅匹配键 |
|
||
| NMT | Network Management | CANopen 网络管理，如远程启动节点 |
|
||
| EDS | Electronic Data Sheet | CANopen 设备对象字典描述文件 |
|
||
| RS-232 / RS-485 | — | 串行通信物理层标准 |
|
||
| OPC UA | Open Platform Communications Unified Architecture | 北向工业互操作标准，默认端口 4840 |
|
||
| open62541 | — | 开源 OPC UA 协议栈，本项目北向实现基础 |
|
||
| Framer / 分帧器 | IProtocolFramer | 插件 Stage 1：从字节流提取完整协议帧 |
|
||
| Parser / 解析器 | IProtocolParser | 插件 Stage 2：将完整帧解析为 DOMMessage |
|
||
| Plugin ABI | — | 插件 C ABI：`create_framer` / `create_parser` |
|
||
| protocol_modbus.so | — | Modbus 协议动态库插件 |
|
||
| protocol_canopen.so | — | CANopen 协议动态库插件 |
|
||
|
||
#### 0.1.6 Edge 采集子系统
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| Ingress | IIngressPort | 边缘采集入口抽象：ModbusIngress、CanIngress、EthIngress |
|
||
| Egress | — | 边缘写出路径：ModbusEgress、CanEgress、EthEgress |
|
||
| CompositeIngress | — | 组合多个协议 Ingress 的统一 poll 入口 |
|
||
| IngressWorker | — | 独立采集线程，循环 poll 并将结果推入 PacketQueue |
|
||
| PacketQueue | — | 线程安全 RawPacket 队列，主线程 drain 后上行发送 |
|
||
| ChannelManager | — | 物理通道管理器族：Modbus / CAN / Eth |
|
||
| ioMutex | — | 每物理通道独立互斥锁，保护采集线程与下行写并发 |
|
||
| ModbusPollScheduler | — | Modbus 批量读计划器，合并相邻寄存器 |
|
||
| PollPlanner | — | 按 pollIntervalMs 调度到期测点 |
|
||
| ChangeFilter | — | 工程值变化与死区过滤 |
|
||
| PdoMapper | — | CANopen PDO 帧与物理点映射 |
|
||
| SdoClient | — | CANopen SDO 写客户端 |
|
||
| NmtManager | — | CANopen 远程节点启动管理 |
|
||
| CommandExecutor | — | edge 下行命令执行器，按 protocol 分发至 Egress |
|
||
| ConfigWatcher | — | 配置文件变更监视，支持热加载 |
|
||
| PhysicalRegistry | — | edge 物理通道与设备注册表 |
|
||
| Mock 模式 | --mock | 无硬件时模拟 Modbus 帧上行 |
|
||
|
||
#### 0.1.7 Daemon 处理与北向
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| Pipeline / PipelineEngine | — | RawPacket → 分帧 → 解析 → 绑定 → Bus/OPC UA 的处理管道 |
|
||
| PluginLoader | — | 动态加载协议插件.so |
|
||
| PointRegistry | — | 逻辑点表与 ObjectModelTree 加载索引 |
|
||
| BindingRegistry | — | 全局/本地绑定加载与 physical ↔ logical 解析 |
|
||
| CommandService | — | 下行命令提交、路由、ACK 等待与状态机 |
|
||
| CommandDispatcher | — | Phase 1 遗留命令转发组件（已由 CommandService 扩展） |
|
||
| HeartbeatMonitor | — | 订阅 heartbeat 主题，超时判定 edge offline |
|
||
| StorageService | — | 订阅 Bus 主题，时序/命令/事件 JSONL 落盘 |
|
||
| ProfileLoader | — | 加载 daemon_profile.json 运行配置 |
|
||
| UaServerEngine | — | OPC UA 服务端引擎 |
|
||
| UaAddressSpaceBuilder | — | 由 ObjectModelTree 动态构建 OPC UA 节点树 |
|
||
| UaModelBinder | — | DOMMessage / SoftbusDataPoint → OPC UA Variable 写值 |
|
||
| UaMethodBinder | — | OPC UA ExecuteCommand Method 入口绑定 |
|
||
| UaEventBridge | — | Bus 事件 → OPC UA 事件桥接（Phase 1 stub） |
|
||
| Variable | — | OPC UA 变量节点，暴露测点当前值 |
|
||
| Method | — | OPC UA 方法节点，用于 ExecuteCommand 下行 |
|
||
| MonitoredItem | — | OPC UA 订阅监控项（Phase 2 规划完整实现） |
|
||
| Subscription | — | OPC UA 客户端数据变更订阅机制 |
|
||
|
||
#### 0.1.8 数据服务（Phase 2～4 规划）
|
||
|
||
| 术语 / 缩写 | 英文 / 标识 | 说明 |
|
||
|-------------|-------------|------|
|
||
| 数据管理 | Data Management | 元数据、质量码、生命周期、存储策略、访问控制、血缘 |
|
||
| 数据订阅 | Data Subscription | 按主题/逻辑点/对象路径过滤推送；Bus 订阅 + OPC UA Subscription |
|
||
| 数据融合 | Data Fusion | 多源时序对齐、关联计算、合成测点发布 |
|
||
| 数据恢复 | Data Recovery | 断线补传、历史回放、缺测插值、冗余源切换 |
|
||
| 数据服务层 | Data Service Layer | 叠加于采集解析层之上的 Bus 中心化能力层 |
|
||
| MetadataRegistry | — | 元数据注册表（代码 stub，待接入 CoreService） |
|
||
| CatalogService | — | 数据资产编目与检索（规划） |
|
||
| RetentionPolicy | — | 时序/事件数据保留与归档策略（规划） |
|
||
| SubscriptionManager | — | 统一订阅注册、QoS、背压管理（规划） |
|
||
| UaSubscriptionBridge | — | Bus DataPoint 驱动 OPC UA MonitoredItem（规划） |
|
||
| SubGateway | — | 北向 MQTT/HTTP 订阅网关（规划） |
|
||
| FusionEngine | — | 融合规则执行与合成测点发布（规划） |
|
||
| FusionRuleRegistry | — | 融合规则配置：输入点、表达式、对齐窗口（规划） |
|
||
| TimeAligner | softbus_core::TimeAligner | 多通道时序对齐器，支持 Nearest/Linear 策略（已实现，待接入） |
|
||
| AlignedSample | — | 对齐后的多通道样本集合，含 anchorTimestampNs |
|
||
| anchor / 锚点时刻 | — | 融合对齐的时间基准点 |
|
||
| SyntheticPoint / 合成测点 | — | 由多输入计算得到的衍生 logicalPoint |
|
||
| RecoveryService | — | daemon 侧缺口检测、回放、冗余切换（规划） |
|
||
| RecoveryBuffer | — | edge 断线期间 RawPacket 环形/持久缓冲（规划） |
|
||
| Gap / 数据缺口 | GapReport | 通信中断导致的时间区间 `[t0, t1]` |
|
||
| replayBatch | — | uplink 补传批量 RawPacket 的扩展消息类型（规划） |
|
||
| QoS | Quality of Service | 订阅侧采样率、可靠性、延迟上限等服务质量策略 |
|
||
| 血缘 | Data Lineage | physicalPoint → binding → logicalPoint → fusedPoint 溯源链 |
|
||
| ACL | Access Control List | 读/写/订阅权限控制（规划） |
|
||
|
||
#### 0.1.9 缩写词速查
|
||
|
||
| 缩写 | 全称 |
|
||
|------|------|
|
||
| AI / AO | Analog Input / Analog Output（模拟量输入/输出，signalType 字段） |
|
||
| ABI | Application Binary Interface |
|
||
| CAN | Controller Area Network |
|
||
| DOM | Document Object Model |
|
||
| IPC | Inter-Process Communication（如 D-Bus，Phase 2 规划） |
|
||
| JSON | JavaScript Object Notation |
|
||
| MES / SCADA | 制造执行系统 / 数据采集与监控系统 |
|
||
| NMT | Network Management（CANopen） |
|
||
| OPC | Open Platform Communications |
|
||
| PDO / SDO | Process / Service Data Object（CANopen） |
|
||
| QoS | Quality of Service |
|
||
| RTU | Remote Terminal Unit（Modbus 串口从站通信模式） |
|
||
| SLA | Service Level Agreement |
|
||
| TCP | Transmission Control Protocol |
|
||
| TLS | Transport Layer Security |
|
||
| UA | Unified Architecture（OPC UA） |
|
||
|
||
---
|
||
|
||
### 0.2 软总线需求
|
||
|
||
针对**系统运行过程中，设备数目灵活增加、供求关系复杂多变，同一内涵的数据在不同设备之间的数据结构、映射关系、传输协议等外在表现有较大差异，导致数据管理与通信困难。与此同时在数据传输过程中，现有传感系统仅起到数据交换的作用，并未发挥其在数据传递过程中信息管理的作用。**
|
||
|
||
为实现多源异构信息设备的互联互通，亟需开发一种软总线架构，用以实现大量异构网络之间的灵活通信。该软总线能够基于设备运行数据构建整体**系统传感软件子系统，并动态分析设备间的潜在关联，促进设备感知信息互补与数据恢复，最终支撑稳定、可靠、高效和低时延的通信传感网络形成。
|
||
|
||
SoftBus OPC 是本需求在工业自动化场景下的 **Phase 1 原型实现**：以南向多协议采集、中间层统一语义建模、北向 OPC UA 暴露为核心路径，验证“物理域—逻辑域”双域分离与可扩展插件化解析架构。
|
||
|
||
**能力定位说明：** 当前工程已完成“**数据采集与透传**”主链路（现场 → edge → daemon → OPC UA / JSONL 落盘），尚未形成完整的数据管理、数据恢复、数据订阅与数据融合子系统。第 0.2 节背景中提出的“信息管理、关联分析、互补与恢复”等目标，将在 Phase 2～4 基于现有 Bus 与语义模型逐层叠加（详见第 4 章）。
|
||
|
||
#### 表 0-3 能力演进对照（背景需求 → 软件能力）
|
||
|
||
| 背景需求 | Phase 1 现状 | 目标能力 |
|
||
|----------|--------------|----------|
|
||
| 异构设备互联互通 | edge 多协议采集 + 插件解析 | 已具备基础 |
|
||
| 数据传递过程中的**信息管理** | Binding + 逻辑点表 + 主题路由 | **数据管理**（第 4.6 节） |
|
||
| 设备感知信息**互补** | 单点上报，无跨源合成 | **数据融合**（第 4.4 节） |
|
||
| **数据恢复** | 仅 JSONL 追加写，无回放/补洞 | **数据恢复**（第 4.5 节） |
|
||
| 稳定低时延通信网络 | 多线程采集 + 心跳监测 | 订阅 QoS + 冗余（第 4.3 节） |
|
||
|
||
#### 表 0-4 目前已有的协议与接口
|
||
|
||
| 数据接口类型 | 数据采集方式 | 物理接口类型 | 本系统支持情况 |
|
||
|------------------|---------------------|----------------|-----------------------------------------|
|
||
| 电流模拟量数据 | 模拟传输 | 板卡端子块 | 经 Modbus/OPC 间接接入（Phase 1） |
|
||
| 电压模拟量数据 | 模拟传输 | 板卡端子块 | 经 Modbus/OPC 间接接入（Phase 1） |
|
||
| 串口数据 | RS-232 / RS-485 | DB / 端子块 | **已支持**（libmodbus RTU） |
|
||
| 以太网数据 | TCP/IP 传输 | RJ45 | **已支持**（Modbus TCP、RTU-over-TCP） |
|
||
| CAN 数据 | CAN 2.0A/B 传输 | CAN-H / CAN-L | **已支持**（SocketCAN、CANopen PDO/SDO）|
|
||
|
||
---
|
||
|
||
## 第 1 章 软件总体范围
|
||
|
||
本文档旨在促进多源异构设备的信息互联交换，适用于工业自动化及其他需要统一数据语义与北向开放接口的领域。
|
||
|
||
### 1.1 文档范围
|
||
|
||
| 范围项 | 说明 |
|
||
|------------------|----------------------------------------------------------------------|
|
||
| 软件形态 | 守护进程 `softbus_daemon`、边缘代理 `softbus_edge`、协议插件动态库 |
|
||
| 通信南向 | Modbus RTU/TCP、CANopen、裸以太网（骨架） |
|
||
| 通信北向 | OPC UA（open62541），端口默认 4840 |
|
||
| 边缘—中心链路 | TCP JSON 行协议（UplinkWire），端口默认 9000 |
|
||
| 配置与模型 | 物理层配置、逻辑点表、全局/本地绑定、对象模型五层寻址 |
|
||
| **Phase 1 已交付** | 数据采集、协议解析、OPC UA 读写、命令下行、JSONL 时序落盘、Bus 主题发布 |
|
||
| **Phase 2～4 规划** | 数据管理、数据订阅、数据融合、数据恢复（见第 4 章） |
|
||
| 不在 Phase 1 范围 | D-Bus 主流程接入、多 edge 负载均衡、TLS 生产部署、时序数据库集群 |
|
||
|
||
### 1.2 交付物
|
||
|
||
```
|
||
build/bin/softbus_daemon # 守护进程
|
||
build/bin/softbus_edge # 边缘代理
|
||
build/lib/protocol_modbus.so # Modbus 协议插件
|
||
build/lib/protocol_canopen.so # CANopen 协议插件
|
||
config/ # 运行时配置
|
||
docs/ # 架构与模块文档
|
||
```
|
||
|
||
---
|
||
|
||
## 第 2 章 总体概述
|
||
|
||
### 2.1 设计目标
|
||
|
||
SoftBus OPC 的总体目标分两个阶段理解：
|
||
|
||
**Phase 1（当前）：采集与透传** — 将工业现场总线数据接入软总线，经统一对象模型暴露为 OPC UA 地址空间，并完成基础落盘。
|
||
|
||
**Phase 2～4（演进）：数据服务** — 在采集层之上构建数据管理、订阅、融合与恢复能力，使软总线从“数据管道”升级为“数据服务中枢”，回应第 0.2 节背景中对信息管理、互补与恢复的要求。
|
||
|
||
Phase 1 已实现：
|
||
|
||
1. **南向统一接入**：边缘代理按物理配置轮询/订阅多协议设备，产出标准化 `RawPacket`。
|
||
2. **中间层协议中立**：Daemon 通过插件分帧/解析，不侵入核心数据模型。
|
||
3. **北向标准开放**：OPC UA 客户端读写 Variable、调用 Method 下发控制命令。
|
||
4. **双域解耦**：物理配置在 edge，逻辑模型与绑定在 daemon，支持独立演进与热加载。
|
||
|
||
Phase 2～4 待建设（详见第 4 章）：
|
||
|
||
5. **数据管理**：统一元数据、质量码、保留策略与访问视图。
|
||
6. **数据订阅**：多消费者按主题/QoS 订阅，不仅限于 OPC UA 轮询读。
|
||
7. **数据融合**：多 edge、多协议测点在时间轴上对齐并合成衍生量。
|
||
8. **数据恢复**：通信中断后的补传、历史回填与冗余切换。
|
||
|
||
### 2.2 系统分层架构
|
||
|
||
```
|
||
L3 可执行层
|
||
softbus_daemon ─────────────────────────────┐
|
||
softbus_edge │
|
||
│
|
||
L2 服务/协议层 │
|
||
softbus_opcua · softbus_transport │
|
||
softbus_pipeline · softbus_command │
|
||
softbus_monitor · softbus_storage │
|
||
softbus_registry · softbus_bus │
|
||
plugins/protocol_modbus · protocol_canopen │
|
||
[L2+ 规划] 数据订阅/融合/恢复/管理子系统 │
|
||
│
|
||
L1 契约层 │
|
||
softbus_api (ExecuteCommand) │
|
||
│
|
||
L0 内核层 │
|
||
softbus_core ◄──────────────────────────────┘
|
||
（模型、日志、插件 ABI、配置、身份标识）
|
||
```
|
||
|
||
#### 表 2-1 模块职责速查
|
||
|
||
| 模块 | 路径 | 职责 |
|
||
|--------------------|---------------------------|------------------------------------------------|
|
||
| softbus_core | src/softbus_core | 数据模型、Plugin ABI、日志、配置、PlatformSignal |
|
||
| softbus_api | src/softbus_api | ExecuteCommand 等跨模块控制面契约 |
|
||
| softbus_transport | src/softbus_transport | TcpUplinkClient/Server、UplinkWire 编解码 |
|
||
| softbus_registry | src/softbus_registry | 逻辑点表、BindingRegistry、PointRegistry |
|
||
| softbus_pipeline | src/softbus_pipeline | RawPacket 解析管道与 Bus 订阅 |
|
||
| softbus_opcua | src/softbus_opcua | OPC UA 服务端、地址空间、Method、事件桥 |
|
||
| softbus_command | src/softbus_command | 下行命令提交、状态机、ACK 等待 |
|
||
| softbus_monitor | src/softbus_monitor | 心跳超时监测、edge 在线/离线事件 |
|
||
| softbus_storage | src/softbus_storage | 时序数据落盘（JSONL） |
|
||
| softbus_bus | src/softbus_bus | 进程内 Pub/Sub 消息总线 |
|
||
| softbus_edge | src/softbus_edge | 物理采集、上行、下行写回 |
|
||
| softbus_daemon | src/softbus_daemon | 编排中心（CoreService） |
|
||
| plugins | src/plugins | Modbus / CANopen 协议动态库 |
|
||
|
||
### 2.3 运行时拓扑
|
||
|
||
```
|
||
OPC UA Client 现场设备
|
||
opc.tcp://:4840 Modbus / CAN / Eth
|
||
│ ▲
|
||
▼ │ 驱动 I/O
|
||
┌──────────────────────────────────────────────────────────┐
|
||
│ softbus_daemon │
|
||
│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ │
|
||
│ │ UaServer │◄──│ UaModelBinder│◄──│ PipelineEngine│ │
|
||
│ │ Engine │ └──────────────┘ └───────▲───────┘ │
|
||
│ └──────┬──────┘ │ │
|
||
│ │ ExecuteCommand │ RawPacket│
|
||
│ ┌──────▼──────┐ ┌──────────────┐ ┌───────┴───────┐ │
|
||
│ │CommandService│ │ Bus (Pub/Sub)│◄──│TransportBridge│ │
|
||
│ └──────┬──────┘ └──────────────┘ └───────▲───────┘ │
|
||
│ │ ┌─────────┴────────┐ │
|
||
│ ┌──────▼──────────────────────────│ TcpUplinkServer │ │
|
||
│ │ HeartbeatMonitor · StorageService│ :9000 │ │
|
||
│ └──────────────────────────────────┴────────┬────────┘ │
|
||
└───────────────────────────────────────────────┼──────────┘
|
||
│ TCP JSON
|
||
┌───────────────────────────────────────────────▼──────────┐
|
||
│ softbus_edge │
|
||
│ ┌─────────────────┐ ┌─────────────────────────┐ │
|
||
│ │ IngressWorker │ poll │ CompositeIngress │ │
|
||
│ │ (采集线程) │────────►│ Modbus / CAN / Eth │ │
|
||
│ └────────┬────────┘ └─────────────────────────┘ │
|
||
│ │ push │
|
||
│ ┌────────▼────────┐ drain ┌─────────────────────────┐ │
|
||
│ │ PacketQueue │──────────►│ 主线程：心跳 + uplink │ │
|
||
│ └─────────────────┘ │ CommandExecutor 写回 │ │
|
||
│ └─────────────────────────┘ │
|
||
└──────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
### 2.4 双域配置模型
|
||
|
||
系统采用 **物理域（edge）** 与 **逻辑域（daemon）** 分离：
|
||
|
||
| 配置域 | 典型文件 | 加载方 | 用途 |
|
||
|----------|---------------------------------------|--------|------------------------------------|
|
||
| 物理域 | config/edge/physical_channels.json | edge | 串口、TCP、CAN 接口参数 |
|
||
| 物理域 | config/edge/physical_devices.json | edge | 从站、寄存器、物理点、CANopen 节点 |
|
||
| 物理域 | config/edge/bindings_local.json | edge | 下行写命令本地备用解析（可选） |
|
||
| 逻辑域 | config/daemon/logical_points.json | daemon | OPC UA 逻辑点与对象模型 |
|
||
| 逻辑域 | config/daemon/bindings_global.json | daemon | 物理点 ↔ 逻辑点 ↔ edgeId 绑定 |
|
||
| 运行配置 | config/daemon/daemon_profile.json | daemon | OPC UA 端口、uplink 端口、插件名 |
|
||
| 运行配置 | config/edge/edge.json | edge | edgeId、daemon 地址、心跳周期 |
|
||
|
||
edge 上行时**不做** binding 初筛：所有已配置物理点均采集上报；是否映射到 OPC UA 节点由 daemon 的 `bindings_global.json` 决定。未绑定点进入 `edge/unbound/raw/...` 主题，不进入 Pipeline 解析。
|
||
|
||
### 2.5 对象模型五层结构
|
||
|
||
对象寻址路径格式：`Site/System/Asset/Device/Point`
|
||
|
||
```
|
||
DemoSite
|
||
└── DemoSystem
|
||
└── DemoAsset
|
||
└── EdgeDevice_01
|
||
├── MotorTemperature ← OPC UA Variable
|
||
├── MotorSpeed
|
||
└── ValveOpening ← 可写，支持 Method 下行
|
||
```
|
||
|
||
逻辑点表由 `PointRegistry` 加载，驱动 `UaAddressSpaceBuilder` 动态构建 OPC UA 地址空间。
|
||
|
||
### 2.6 数据流概览
|
||
|
||
#### 2.6.1 上行（设备 → OPC UA）
|
||
|
||
```
|
||
现场硬件
|
||
→ ChannelManager（Modbus/CAN/Eth，per-channel 互斥）
|
||
→ *Ingress::poll()（IngressWorker 采集线程）
|
||
→ PacketQueue
|
||
→ 主线程 drain → UplinkClient::sendRawPacket()
|
||
→ TCP :9000 → TransportBusBridge → Bus
|
||
→ PipelineEngine::process()
|
||
→ protocol_*.so Framer → Parser → DOMMessage
|
||
→ BindingRegistry 查逻辑点
|
||
→ UaModelBinder → OPC UA Variable 写值
|
||
→ StorageService 落盘 timeseries.jsonl
|
||
```
|
||
|
||
#### 2.6.2 下行（OPC UA → 设备）
|
||
|
||
```
|
||
OPC UA Client 调用 ExecuteCommand Method
|
||
→ UaMethodBinder
|
||
→ CommandService::submit()
|
||
→ BindingRegistry 解析 logicalPointId → physicalPointId + edgeId
|
||
→ TcpUplinkServer::sendCommand()
|
||
→ edge TcpUplinkClient（io 线程回调）
|
||
→ CommandExecutor::execute()
|
||
→ ModbusEgress / CanEgress / EthEgress
|
||
→ ChannelManager 写寄存器 / SDO / 裸帧
|
||
→ SoftbusAck 回传 daemon
|
||
```
|
||
|
||
#### 2.6.3 心跳与在线监测
|
||
|
||
- edge 主线程按 `heartbeatPeriodMs`（默认 1000 ms）向 daemon 发送心跳。
|
||
- daemon `HeartbeatMonitor` 订阅 Bus 心跳主题，超时（默认 5 s）发布 edge offline 事件。
|
||
- 采集与 Modbus 重连已移至 **IngressWorker 独立线程**，主线程不再被阻塞，避免心跳饿死导致误判断线。
|
||
|
||
### 2.7 Edge 多线程架构（已实现）
|
||
|
||
| 线程 | 职责 |
|
||
|-----------------|------------------------------------------------|
|
||
| 主线程 | ConfigWatcher、心跳、tickUplink、drain 队列发包 |
|
||
| IngressWorker | ingress_->poll() → PacketQueue.push() |
|
||
| Uplink ioThread | 异步读 TCP、触发 CommandExecutor 下行写 |
|
||
|
||
并发保护：`ModbusChannelManager`、`CanChannelManager`、`EthChannelManager` 均为每个物理通道维护独立 `ioMutex`，采集线程与 uplink io 线程上的写操作互斥访问同一总线/套接字。
|
||
|
||
### 2.8 全局约束原则
|
||
|
||
1. **传输中立**：`softbus_core` 模型字段不得编码 OPC UA、HTTP 等特定语义。
|
||
2. **单向依赖**：上层可依赖下层，下层不得反向引用 daemon/edge/opcua。
|
||
3. **插件 ABI 稳定**：C ABI 导出 `create_framer` / `create_parser`，保证插件可独立升级。
|
||
4. **traceId 贯穿**：RawPacket → DOMMessage → OPC UA → 日志，全程可追踪。
|
||
5. **热加载**：edge 监视 physical 配置；daemon 监视 logical_points 与 bindings_global（SIGHUP / ConfigWatcher）。
|
||
|
||
---
|
||
|
||
## 第 3 章 Daemon 软件
|
||
|
||
### 3.1 Daemon 概览
|
||
|
||
Daemon 软件（`softbus_daemon`）作为软总线软件的**核心运行进程**，实现以下功能：
|
||
|
||
- 监听边缘代理 TCP 上行，接收 `RawPacket`、心跳、注册信息；
|
||
- 加载协议插件，将原始帧解析为统一语义消息；
|
||
- 维护逻辑点表与全局绑定，驱动 OPC UA 地址空间；
|
||
- 接收北向 OPC UA Method 调用，经 CommandService 路由至对应 edge；
|
||
- 监测 edge 心跳，发布在线/离线事件；
|
||
- 可选时序数据持久化。
|
||
|
||
Daemon **不**直接访问串口、CAN 或 Modbus 硬件；所有物理 I/O 由 edge 完成。
|
||
|
||
---
|
||
|
||
### 3.2 架构设计
|
||
|
||
#### 3.2.1 组件关系
|
||
|
||
`CoreService` 是唯一编排入口，按序装配各子系统：
|
||
|
||
```
|
||
CoreService（编排中心）
|
||
├── ProfileLoader ← daemon_profile.json
|
||
├── PointRegistry ← logical_points.json / topic_rule.json
|
||
├── BindingRegistry ← bindings_global.json
|
||
├── PluginLoader (modbus) ← protocol_modbus.so
|
||
├── Bus::instance() ← 进程内消息总线
|
||
├── CommandService ← 下行命令状态机
|
||
├── StorageService ← 时序落盘
|
||
├── HeartbeatMonitor ← edge 心跳超时
|
||
├── UaServerEngine ← OPC UA 服务端 :4840
|
||
├── UaAddressSpaceBuilder ← 由 ObjectModelTree 构建节点
|
||
├── UaModelBinder ← DOMMessage → Variable 写值
|
||
├── UaMethodBinder ← ExecuteCommand Method 入口
|
||
├── UaEventBridge ← 事件桥接（Phase 1 stub）
|
||
├── PipelineEngine ← RawPacket 解析管道
|
||
├── TransportBusBridge ← TCP ↔ Bus 桥接
|
||
├── TcpUplinkServer ← 上行监听 :9000
|
||
└── ConfigWatcher ← 配置热加载
|
||
```
|
||
|
||
#### 3.2.2 线程模型
|
||
|
||
| 线程 / 上下文 | 职责 |
|
||
|--------------------|---------------------------------------------------|
|
||
| 主线程（run 循环） | uaEngine_->iterate()、heartbeatMonitor_.tick()、ConfigWatcher |
|
||
| ioThread_ | asio io_context，处理 TcpUplinkServer 异步 I/O |
|
||
| OPC UA 内部线程 | open62541 服务端事件（由 iterate 驱动） |
|
||
|
||
主循环伪代码：
|
||
|
||
```cpp
|
||
while (!shouldStop()) {
|
||
configWatcher_.poll();
|
||
if (reloadRequested()) reloadConfig();
|
||
uaEngine_->iterate();
|
||
heartbeatMonitor_.tick();
|
||
sleep(10ms);
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
### 3.3 功能设计
|
||
|
||
#### 3.3.1 上行接入
|
||
|
||
| 功能项 | 实现组件 | 说明 |
|
||
|--------------|-----------------------|------------------------------------------------|
|
||
| TCP 监听 | TcpUplinkServer | 默认 127.0.0.1:9000，支持多 edge 连接 |
|
||
| 线协议解码 | UplinkWire | JSON 行：rawPacket / executeCommand / heartbeat |
|
||
| Bus 发布 | TransportBusBridge | 按 topic 规则发布至 Bus |
|
||
| 协议解析 | PipelineEngine + 插件 | Framer → Parser → DOMMessage |
|
||
| 绑定映射 | BindingRegistry | physicalPointId → logicalPointId |
|
||
| OPC 写值 | UaModelBinder | 更新 Variable 节点 |
|
||
| 时序存储 | StorageService | 写入 data/logs/timeseries.jsonl |
|
||
|
||
#### 3.3.2 下行控制
|
||
|
||
| 功能项 | 实现组件 | 说明 |
|
||
|--------------|-----------------------|------------------------------------------------|
|
||
| Method 入口 | UaMethodBinder | OPC UA ExecuteCommand 方法 |
|
||
| 命令校验 | CommandService | objectRef 校验、路由解析、状态机 |
|
||
| edge 路由 | BindingRegistry | logicalPointId → edgeId + physicalPointId |
|
||
| TCP 发送 | TcpUplinkServer | 编码 ExecuteCommand 线消息 |
|
||
| ACK 等待 | CommandService | 等待 edge SoftbusAck，返回 JSON 结果 |
|
||
|
||
#### 3.3.3 运维与监测
|
||
|
||
| 功能项 | 实现组件 | 说明 |
|
||
|--------------|-----------------------|------------------------------------------------|
|
||
| 心跳监测 | HeartbeatMonitor | 超时判定 edge offline |
|
||
| 配置热加载 | ConfigWatcher + SIGHUP| 重载 logical_points、bindings_global |
|
||
| 插件热路径 | resolvePluginPath() | 多目录搜索 libprotocol_*.so |
|
||
| 自测模式 | --self-test | 启动 3 s 后自动下发写命令 |
|
||
|
||
---
|
||
|
||
### 3.4 方案设计
|
||
|
||
#### 3.4.1 启动序列
|
||
|
||
`CoreService::start(profilePath, logicalPointsPath, bindingsGlobalPath)` 按序执行：
|
||
|
||
| 步骤 | 操作 | 失败行为 |
|
||
|------|-------------------------------------------|------------|
|
||
| 1 | ProfileLoader 加载 daemon_profile.json | 返回 false |
|
||
| 2 | PointRegistry 加载 logical_points.json | 返回 false |
|
||
| 3 | BindingRegistry 加载 bindings_global.json | 返回 false |
|
||
| 4 | PluginLoader 加载 protocol_modbus.so | 返回 false |
|
||
| 5 | 初始化 Bus、CommandService、Storage、Monitor | — |
|
||
| 6 | UaServerEngine 启动 OPC UA | 返回 false |
|
||
| 7 | UaAddressSpaceBuilder 构建地址空间 | 返回 false |
|
||
| 8 | PipelineEngine 挂载 Bus 订阅 | — |
|
||
| 9 | TcpUplinkServer 启动 + ioThread | — |
|
||
| 10 | UaMethodBinder 绑定 CommandService | — |
|
||
|
||
#### 3.4.2 Pipeline 三阶段处理
|
||
|
||
```
|
||
process(RawPacket)
|
||
Stage 1 分帧 plugin_->framer()->extract(payload)
|
||
Stage 2 解析 plugin_->parser()->parseFrame(frame, sourceId) → DOMMessage
|
||
Stage 3 绑定 bindings 查表 → UaModelBinder::bind(DOMMessage)
|
||
StorageService 记录时序
|
||
```
|
||
|
||
- 分帧失败：使用原始 payload 继续解析；
|
||
- 解析失败：记录 warnTrace 日志，丢弃消息；
|
||
- 无绑定：发布至 unbound 主题，不进入 OPC 绑定。
|
||
|
||
#### 3.4.3 插件路径解析
|
||
|
||
`resolvePluginPath(pluginName, profilePath)` 搜索顺序：
|
||
|
||
1. `{profileDir}/lib/`
|
||
2. `{cwd}/config/lib/`
|
||
3. `{exeDir}/`、`{exeDir}/../lib/`
|
||
4. `build/lib/`、`lib/`
|
||
|
||
#### 3.4.4 CLI 参数
|
||
|
||
| 参数 | 默认值 | 说明 |
|
||
|-----------------------|-------------------------------------|----------------|
|
||
| `--config` | config/daemon/daemon_profile.json | 运行配置 |
|
||
| `--logical-points` | config/daemon/logical_points.json | 逻辑点表 |
|
||
| `--bindings-global` | config/daemon/bindings_global.json | 全局绑定 |
|
||
| `--point-table` | — | 兼容旧参数 |
|
||
| `--self-test` | 关闭 | 自动写值测试 |
|
||
|
||
---
|
||
|
||
### 3.5 模块设计
|
||
|
||
#### 3.5.1 CoreService 核心服务模块
|
||
|
||
**路径：** `src/softbus_daemon/include/softbus_daemon/core/CoreService.h`
|
||
**职责：** Daemon 唯一编排入口，负责子系统生命周期管理、配置加载、主循环驱动。
|
||
|
||
##### （1）类接口
|
||
|
||
| 方法 | 说明 |
|
||
|--------------------------------------------------------------|--------------------------|
|
||
| `start(profilePath, logicalPointsPath, bindingsGlobalPath)` | 按序启动全部子系统 |
|
||
| `stop()` | 停止 uplink、io、OPC UA |
|
||
| `run(selfTest)` | 阻塞主循环直至收到退出信号 |
|
||
|
||
##### （2）核心成员
|
||
|
||
| 成员 | 类型 | 职责 |
|
||
|-----------------------|-------------------------|------------------------------|
|
||
| `registry_` | PointRegistry | 逻辑点表与 ObjectModelTree |
|
||
| `bindingRegistry_` | BindingRegistry | 物理—逻辑绑定与下行路由 |
|
||
| `modbusPlugin_` | PluginLoader | Modbus 协议插件句柄 |
|
||
| `pipeline_` | PipelineEngine | 上行解析管道 |
|
||
| `commandService_` | CommandService | 下行命令服务 |
|
||
| `uaEngine_` | UaServerEngine | OPC UA 服务端 |
|
||
| `uaBinder_` | UaModelBinder | 语义消息 → Variable |
|
||
| `uaMethodBinder_` | UaMethodBinder | Method → CommandService |
|
||
| `uplink_` | TcpUplinkServer | 边缘 TCP 接入 |
|
||
| `transportBridge_` | TransportBusBridge | TCP 与 Bus 双向桥接 |
|
||
| `heartbeatMonitor_` | HeartbeatMonitor | edge 在线状态 |
|
||
| `storageService_` | StorageService | 时序持久化 |
|
||
| `configWatcher_` | ConfigWatcher | 配置文件变更监视 |
|
||
|
||
##### （3）关键回调绑定
|
||
|
||
```cpp
|
||
// 上行：TransportBusBridge 将 RawPacket 发布至 Bus
|
||
// PipelineEngine 订阅 edge/raw/# 并 process()
|
||
|
||
// 下行：OPC UA Method → CommandService
|
||
uaMethodBinder_->bindExecuteCommand([this](const ExecuteCommand& cmd) {
|
||
return commandResultJson(commandService_->submit(cmd, true));
|
||
});
|
||
```
|
||
|
||
##### （4）配置热加载
|
||
|
||
- `SIGHUP` 或 `ConfigWatcher` 触发 `reloadConfig()`；
|
||
- 重载 `logical_points.json`、`bindings_global.json`；
|
||
- 通过 `PointRegistry::diff()` 记录新增逻辑点数量；
|
||
- Phase 1 热加载**不**重建 OPC UA 地址空间（增量构建为 Phase 2 规划）。
|
||
|
||
##### （5）依赖关系
|
||
|
||
```
|
||
CoreService
|
||
→ softbus_core, softbus_api, softbus_transport
|
||
→ softbus_pipeline, softbus_command, softbus_monitor
|
||
→ softbus_storage, softbus_registry, softbus_bus
|
||
→ softbus_opcua, asio
|
||
→ protocol_modbus（构建依赖）
|
||
```
|
||
|
||
---
|
||
|
||
#### 3.5.2 其他 Daemon 关联模块（概要）
|
||
|
||
| 模块 | 路径 | 与 CoreService 关系 |
|
||
|-------------------|------------------------------|----------------------------------------|
|
||
| PipelineEngine | src/softbus_pipeline | CoreService 创建并 attachBus |
|
||
| CommandService | src/softbus_command | CoreService 创建，MethodBinder 调用 |
|
||
| HeartbeatMonitor | src/softbus_monitor | CoreService 创建，主循环 tick |
|
||
| StorageService | src/softbus_storage | CoreService 创建，attachBus 落盘 |
|
||
| TransportBusBridge| src/softbus_transport | CoreService 创建，连接 uplink 与 Bus |
|
||
| PointRegistry | src/softbus_registry | CoreService 加载逻辑配置 |
|
||
| BindingRegistry | src/softbus_registry | CoreService 加载全局绑定 |
|
||
| UaServerEngine | src/softbus_opcua | CoreService 启动北向服务 |
|
||
|
||
---
|
||
|
||
## 第 4 章 数据服务层演进（采集之上的能力扩展）
|
||
|
||
> **本章定位：** 说明当前架构仅覆盖“采集—解析—暴露—落盘”主链路；数据管理、数据恢复、数据订阅、数据融合将在同一 Bus 与语义模型之上分层建设，而非替换现有 edge/daemon 分工。
|
||
|
||
### 4.1 当前能力边界（Phase 1）
|
||
|
||
#### 已具备
|
||
|
||
| 能力 | 实现位置 | 说明 |
|
||
|----------------|----------------------------------|------------------------------------------------|
|
||
| 多协议采集 | softbus_edge | Modbus / CANopen / Eth，IngressWorker 多线程 |
|
||
| 协议解析 | PipelineEngine + plugins | RawPacket → DOMMessage |
|
||
| 物理—逻辑映射 | BindingRegistry | physicalPointId → logicalPointId |
|
||
| 北向读 | UaModelBinder | DOMMessage → OPC UA Variable（最新值） |
|
||
| 北向写/命令 | CommandService | OPC UA Method → edge 写回 |
|
||
| 进程内发布 | Bus `data/{deviceId}/{pointId}` | Pipeline 解析后发布 SoftbusDataPoint |
|
||
| 基础落盘 | StorageService | 订阅 `data/#` 写入 timeseries.jsonl |
|
||
| 在线监测 | HeartbeatMonitor | edge 心跳超时 → offline 事件 |
|
||
| 变化过滤 | edge ChangeFilter | deadband / reportOnChange，减少上行流量 |
|
||
|
||
#### 尚未具备
|
||
|
||
| 能力 | 差距说明 |
|
||
|----------------|------------------------------------------------------------------|
|
||
| 数据管理 | 无统一元数据注册表、无质量码策略、无保留/归档策略、无访问 ACL |
|
||
| 数据订阅 | Bus 仅进程内；OPC UA Subscription 未完整实现；无外部 MQTT/HTTP 订阅网关 |
|
||
| 数据融合 | TimeAligner 已实现但未接入 Pipeline；无合成测点/关联规则引擎 |
|
||
| 数据恢复 | 无 edge 断线缓存与补传；无历史回放 API；无冗余源自动切换 |
|
||
|
||
### 4.2 目标总体架构（采集 + 数据服务）
|
||
|
||
在 Phase 1 的“采集管道”之上，增加 **数据服务层（Data Service Layer）**，与采集层通过 Bus 解耦：
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────────────────┐
|
||
│ 北向消费层 (Northbound) │
|
||
│ OPC UA Client · 订阅网关(MQTT/HTTP) · 可视化 · 第三方 MES/SCADA │
|
||
└───────────────────────────────┬─────────────────────────────────────────┘
|
||
│ 订阅 / 查询 / 回放
|
||
┌───────────────────────────────▼─────────────────────────────────────────┐
|
||
│ 数据服务层 (Phase 2～4，待建设) │
|
||
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
|
||
│ │ 数据管理 │ │ 数据订阅 │ │ 数据融合 │ │ 数据恢复 │ │
|
||
│ │ MetadataReg │ │ SubManager │ │ FusionEngine│ │ RecoverySvc │ │
|
||
│ │ Catalog/ACL │ │ QoS/Filter │ │ TimeAligner │ │ Buffer/Replay│ │
|
||
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
|
||
│ └────────────────┴──────────────┴──────────────┘ │
|
||
│ softbus_bus │
|
||
│ data/# · fused/# · event/# · recovery/# · metadata/# │
|
||
└───────────────────────────────┬─────────────────────────────────────────┘
|
||
│ 已发布 SoftbusDataPoint / DOMMessage
|
||
┌───────────────────────────────▼─────────────────────────────────────────┐
|
||
│ 采集与解析层 (Phase 1，已实现) │
|
||
│ edge 采集 → uplink → PipelineEngine → Binding → UaModelBinder │
|
||
│ StorageService (JSONL append) │
|
||
└───────────────────────────────┬─────────────────────────────────────────┘
|
||
│
|
||
现场设备 / 传感器
|
||
```
|
||
|
||
**设计原则：**
|
||
|
||
1. **采集不变**：edge 仍只负责物理 I/O 与 RawPacket 上行，不承载融合/恢复逻辑。
|
||
2. **Bus 为中心**：所有数据服务模块仅订阅/发布 Bus 主题，不反向依赖 OPC UA 或 TCP 细节。
|
||
3. **逻辑点为一等公民**：融合结果、恢复补洞、订阅过滤均以 `logicalPointId` / `objectRef` 标识。
|
||
4. **可渐进交付**：四个子能力可独立 Phase 交付，通过 Bus 主题约定保持兼容。
|
||
|
||
### 4.3 数据订阅（Data Subscription）
|
||
|
||
#### 4.3.1 目标
|
||
|
||
让上层应用按**主题、逻辑点、对象路径或表达式**接收数据推送，而不是仅通过 OPC UA 客户端轮询 Variable 最新值。
|
||
|
||
#### 4.3.2 分层设计
|
||
|
||
| 层级 | 组件（规划） | 职责 |
|
||
|----------------|---------------------|------------------------------------------------|
|
||
| 进程内订阅 | Bus（已有） | `subscribe("data/DemoAsset/#")` 过滤 |
|
||
| 订阅管理 | SubscriptionManager | 注册订阅者、QoS、采样/聚合策略、背压 |
|
||
| OPC UA 订阅 | UaSubscriptionBridge| 将 Bus DataPoint 驱动 OPC UA MonitoredItem |
|
||
| 北向网关 | SubGateway (MQTT等) | 对外发布 `data/{deviceId}/{pointId}` 映射主题 |
|
||
|
||
#### 4.3.3 主题约定（已有基础）
|
||
|
||
`config/daemon/topic_rule.json` 已定义：
|
||
|
||
```json
|
||
{
|
||
"data": "data/{deviceId}/{pointId}",
|
||
"command": "control/cmd/{deviceId}/{pointId}/set",
|
||
"ack": "control/ack/{deviceId}/{pointId}",
|
||
"event": "event/{deviceId}/{eventType}",
|
||
"heartbeat": "system/heartbeat/{deviceId}"
|
||
}
|
||
```
|
||
|
||
Phase 2 扩展主题（规划）：
|
||
|
||
| 主题模式 | 载荷类型 | 用途 |
|
||
|-----------------------|-----------------|--------------------------|
|
||
| `fused/{logicalPointId}` | SoftbusDataPoint | 融合后的合成测点 |
|
||
| `recovery/{edgeId}/gap` | GapReport | 通信中断区间上报 |
|
||
| `metadata/point/{id}` | PointMetadata | 元数据变更通知 |
|
||
|
||
#### 4.3.4 QoS 维度（规划）
|
||
|
||
- **采样率**：订阅侧 maxRateHz，高于采集率时由 SubscriptionManager 降采样。
|
||
- **可靠性**：AtMostOnce / AtLeastOnce；关键测点走持久化队列。
|
||
- **延迟上限**：融合/订阅链路 SLA 监控，超时打 quality=Uncertain。
|
||
|
||
### 4.4 数据融合（Data Fusion）
|
||
|
||
#### 4.4.1 目标
|
||
|
||
实现第 0.2 节所述“**动态分析设备间潜在关联，促进设备感知信息互补**”：将来自不同 edge、不同协议、不同采样周期的测点，对齐到统一时间轴，计算合成量或关联事件。
|
||
|
||
#### 4.4.2 典型场景
|
||
|
||
| 场景 | 输入 | 输出 |
|
||
|------------------|------------------------------------|------------------------------|
|
||
| 多源对齐 | 温度(Modbus) + 转速(CANopen) | 同一 anchor 时刻的对齐样本 |
|
||
| 冗余互补 | 双 Modbus 读同一物理量 | 择优或加权平均 |
|
||
| 衍生计算 | 电流 + 电压 | 功率（合成 logicalPoint） |
|
||
| 事件关联 | 阀位 + 流量 | 开阀-流量响应事件 |
|
||
|
||
#### 4.4.3 组件设计（规划）
|
||
|
||
```
|
||
Bus data/# ──► FusionEngine
|
||
├── TimeAligner（已有 softbus_core，待接入）
|
||
├── FusionRuleRegistry（规则配置：表达式 / 关联表）
|
||
├── SyntheticPointPublisher
|
||
└── publish → fused/# + StorageService
|
||
```
|
||
|
||
**已有基础：** `softbus_core::TimeAligner` 支持按时间窗口（默认 50 ms）多通道 Nearest/Linear 对齐，当前 **未接入** Pipeline 主路径。
|
||
|
||
**配置示例（规划）：**
|
||
|
||
```json
|
||
{
|
||
"fusionRules": [
|
||
{
|
||
"outputLogicalPointId": "EdgeDevice_01/ShaftPower",
|
||
"inputs": ["EdgeDevice_01/MotorCurrent", "EdgeDevice_01/MotorVoltage"],
|
||
"expression": "current * voltage",
|
||
"alignWindowMs": 100
|
||
}
|
||
]
|
||
}
|
||
```
|
||
|
||
### 4.5 数据恢复（Data Recovery）
|
||
|
||
#### 4.5.1 目标
|
||
|
||
在 edge 断连、网络闪断或单源故障时，保证上层看到的数据**可解释、可补全、可切换**，避免“静默丢数”。
|
||
|
||
#### 4.5.2 恢复策略
|
||
|
||
| 策略 | 触发条件 | 行为（规划） |
|
||
|----------------|-----------------------|---------------------------------------------------|
|
||
| edge 本地缓存 | uplink 不可达 | edge 环形缓冲 RawPacket，恢复后按序补传 |
|
||
| daemon 缺口检测| 心跳超时 / 时间戳跳变 | 标记 `[t0,t1]` 为 Gap，发布 recovery 事件 |
|
||
| 历史回放 | 运维 / 算法重算 | RecoveryService 从时序库按时间范围重放至 Bus |
|
||
| 冗余源切换 | 主 physicalPoint 失效 | Binding 中声明 backup，RecoverySvc 切换数据源 |
|
||
| 插值补洞 | 短缺口 & quality=Bad | FusionEngine 或 RecoverySvc 线性/前值填充（可配置）|
|
||
|
||
#### 4.5.3 与 HeartbeatMonitor 的衔接
|
||
|
||
当前 `HeartbeatMonitor` 在超时后发布 offline 事件；RecoveryService 应订阅该事件，联动：
|
||
|
||
1. 将受影响 logicalPoint 的 quality 置为 Bad/Uncertain；
|
||
2. 若存在冗余 binding，自动切换至 backup physicalPoint；
|
||
3. edge 重连后触发 **gap fill 补传协议**（uplink 扩展消息类型 `replayBatch`）。
|
||
|
||
#### 4.5.4 edge 侧扩展（规划）
|
||
|
||
```
|
||
uplink 断连期间：
|
||
PacketQueue（已有）→ 持久化 RingBuffer（新增 RecoveryBuffer）
|
||
uplink 恢复：
|
||
sendReplayBatch(RawPacket[]) → daemon RecoveryService 去重合并
|
||
```
|
||
|
||
### 4.6 数据管理（Data Management）
|
||
|
||
#### 4.6.1 目标
|
||
|
||
使软总线在传递数据的同时承担**信息管理**职责：谁产生、何时有效、质量如何、保留多久、谁能访问。
|
||
|
||
#### 4.6.2 管理维度
|
||
|
||
| 维度 | 内容（规划） | Phase 1 现状 |
|
||
|--------------|-----------------------------------------------------------|---------------------|
|
||
| 元数据 | 点描述、单位、量程、采样周期、数据源 edgeId | logical_points 部分有 |
|
||
| 质量码 | Good / Uncertain / Bad，带来源与时间戳 | 未映射至 OPC Status |
|
||
| 生命周期 | 热数据 / 温数据 / 归档策略 | 仅 JSONL 无限追加 |
|
||
| 目录与检索 | 按 Site/System/Asset 浏览、按 tag 搜索 | ObjectModelTree 静态 |
|
||
| 访问控制 | 读/写/订阅权限（与 OPC UA Role 或外部 IAM 对接） | 未实现 |
|
||
| 血缘 | physicalPoint → binding → logicalPoint → fusedPoint | 仅 binding 静态配置 |
|
||
|
||
#### 4.6.3 组件设计（规划）
|
||
|
||
```
|
||
MetadataRegistry（daemon 已有 stub，未接入 CoreService）
|
||
├── 与 PointRegistry / BindingRegistry 联合索引
|
||
├── 发布 metadata/# 变更事件
|
||
└── 为 SubscriptionManager / FusionEngine 提供查询 API
|
||
|
||
CatalogService
|
||
├── 数据资产编目（设备、测点、主题、保留策略）
|
||
└── StorageService 升级：按策略轮转 / 压缩 / 导出 Parquet
|
||
|
||
RetentionPolicy
|
||
└── 配置：timeseries 保留 30 天，event 保留 90 天，fused 保留 7 天
|
||
```
|
||
|
||
#### 4.6.4 StorageService 演进
|
||
|
||
当前 `StorageService` 订阅 `data/#`、`control/ack/#`、`event/#` 追加 JSONL。
|
||
|
||
演进方向：
|
||
|
||
- 分表/分文件：按 deviceId 或日期切分；
|
||
- 索引：时间范围查询接口，供 RecoveryService 回放；
|
||
- 与 MetadataRegistry 联动：落盘记录含 quality、sourceEdgeId、bindingId。
|
||
|
||
### 4.7 四能力协作关系
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
subgraph collect [采集层 Phase1]
|
||
Edge[softbus_edge]
|
||
Pipeline[PipelineEngine]
|
||
end
|
||
|
||
subgraph bus [Bus]
|
||
DataTopic[data/#]
|
||
FusedTopic[fused/#]
|
||
RecoveryTopic[recovery/#]
|
||
end
|
||
|
||
subgraph mgmt [数据管理]
|
||
Meta[MetadataRegistry]
|
||
Catalog[CatalogService]
|
||
Storage[StorageService]
|
||
end
|
||
|
||
subgraph sub [数据订阅]
|
||
SubMgr[SubscriptionManager]
|
||
OpcSub[UaSubscriptionBridge]
|
||
Gateway[SubGateway]
|
||
end
|
||
|
||
subgraph fusion [数据融合]
|
||
Align[TimeAligner]
|
||
Fusion[FusionEngine]
|
||
end
|
||
|
||
subgraph recovery [数据恢复]
|
||
RecSvc[RecoveryService]
|
||
EdgeBuf[edge RecoveryBuffer]
|
||
end
|
||
|
||
Edge --> Pipeline
|
||
Pipeline --> DataTopic
|
||
DataTopic --> Storage
|
||
DataTopic --> Align
|
||
Align --> Fusion
|
||
Fusion --> FusedTopic
|
||
DataTopic --> SubMgr
|
||
FusedTopic --> SubMgr
|
||
SubMgr --> OpcSub
|
||
SubMgr --> Gateway
|
||
Meta --> Catalog
|
||
Meta --> SubMgr
|
||
RecSvc --> RecoveryTopic
|
||
EdgeBuf --> RecSvc
|
||
RecSvc --> DataTopic
|
||
HeartbeatMonitor --> RecSvc
|
||
```
|
||
|
||
**典型链路示例：**
|
||
|
||
1. **正常订阅：** 采集 → Pipeline → `data/#` → SubscriptionManager → OPC UA / MQTT 推送。
|
||
2. **融合订阅：** 多路 `data/#` → TimeAligner → FusionEngine → `fused/#` → 订阅者。
|
||
3. **断线恢复：** Heartbeat 超时 → RecoveryService 标记 Gap → edge 补传 → 合并入 Storage → quality 恢复 Good。
|
||
4. **管理审计：** MetadataRegistry 记录血缘 → Catalog 展示 → RetentionPolicy 归档历史。
|
||
|
||
### 4.8 分阶段交付路线
|
||
|
||
| 阶段 | 主题 | 交付内容 | 依赖现有模块 |
|
||
|--------|----------------|--------------------------------------------------------------|---------------------------|
|
||
| Phase 1 | 数据采集 | edge 多线程采集、Pipeline、OPC UA、JSONL 落盘、Bus 发布 | **当前** |
|
||
| Phase 2 | 数据订阅 + 管理 | SubscriptionManager、OPC UA Subscription、MetadataRegistry 接入、Storage 分片与索引 | Bus、PointRegistry、StorageService |
|
||
| Phase 3 | 数据融合 | TimeAligner 接入、FusionRule 配置、合成测点发布至 fused/# | Pipeline publishDataPoint |
|
||
| Phase 4 | 数据恢复 | edge RecoveryBuffer、gap 协议、冗余 binding 切换、历史回放 API | HeartbeatMonitor、Transport |
|
||
|
||
### 4.9 与 CoreService 的集成方式（规划）
|
||
|
||
四个子能力均作为 **CoreService 的可选子模块** 挂载，不改变 edge 进程边界：
|
||
|
||
```cpp
|
||
// 规划：CoreService::start() 扩展
|
||
metadataRegistry_->load(...);
|
||
subscriptionManager_->attachBus(&bus);
|
||
fusionEngine_->attachBus(&bus); // 订阅 data/#，发布 fused/#
|
||
recoveryService_->attachBus(&bus); // 订阅 event/#、heartbeat
|
||
// StorageService 升级为 IDataStore 接口，供 Recovery 查询
|
||
```
|
||
|
||
模块间**仅通过 Bus 主题与 Registry 查询 API 通信**，避免 CoreService 直接调用 Fusion/Recovery 内部细节，保持与 Phase 1 插件化风格一致。
|
||
|
||
---
|
||
|
||
## 附录 A Edge 软件概要
|
||
|
||
> Edge 详细设计见 [softbus_edge.md](softbus_edge.md)。
|
||
|
||
| 项目 | 说明 |
|
||
|------------|--------------------------------------------------------------|
|
||
| 进程 | softbus_edge |
|
||
| 核心类 | EdgeRuntime、IngressWorker、PacketQueue、CommandExecutor |
|
||
| 采集 | CompositeIngress（ModbusIngress + CanIngress + EthIngress） |
|
||
| 调度 | ModbusPollScheduler、PollPlanner、ChangeFilter |
|
||
| CANopen | NmtManager、PdoMapper、SdoClient |
|
||
| 上行 | UplinkClient → TcpUplinkClient（独立 ioThread） |
|
||
| 并发 | 采集线程 + 主线程 + uplink io 线程；per-channel ioMutex |
|
||
|
||
---
|
||
|
||
## 附录 B 文档索引
|
||
|
||
| 文档 | 内容 |
|
||
|------------------------------------|--------------------------|
|
||
| [ARCHITECTURE.md](ARCHITECTURE.md) | 英文分层架构与构建目标 |
|
||
| [softbus_daemon.md](softbus_daemon.md) | Daemon 模块细节 |
|
||
| [softbus_edge.md](softbus_edge.md) | Edge 物理层与多线程 |
|
||
| [数据流转.md](数据流转.md) | 上下行字段映射 |
|
||
| [UPLINK_WIRE_PROTOCOL.md](UPLINK_WIRE_PROTOCOL.md) | 线协议格式 |
|
||
| [command_flow.md](command_flow.md) | 下行命令时序 |
|
||
| [plugins.md](plugins.md) | 协议插件 ABI |
|
||
|
||
---
|
||
|
||
## 附录 C 演进路线汇总
|
||
|
||
### C.1 Phase 1（当前）— 数据采集与透传
|
||
|
||
- edge 多协议采集（Modbus / CANopen / Eth）+ IngressWorker 多线程；
|
||
- Pipeline 解析 + Binding + OPC UA 暴露；
|
||
- Bus 发布 `data/#` + StorageService JSONL 落盘；
|
||
- CommandService 下行写回 + HeartbeatMonitor 在线监测。
|
||
|
||
### C.2 Phase 2 — 数据订阅与数据管理
|
||
|
||
- SubscriptionManager + OPC UA MonitoredItem/Subscription 完整实现；
|
||
- MetadataRegistry 接入 CoreService，质量码映射至 OPC StatusCode；
|
||
- StorageService 分片、时间范围查询、保留策略；
|
||
- 可选北向 SubGateway（MQTT/HTTP）。
|
||
|
||
### C.3 Phase 3 — 数据融合
|
||
|
||
- TimeAligner 接入 FusionEngine；
|
||
- fusionRules 配置化（表达式 / 关联规则）；
|
||
- 合成测点发布至 `fused/#` 并进入 OPC UA 地址空间；
|
||
- CANopen 插件多路由并行解析。
|
||
|
||
### C.4 Phase 4 — 数据恢复
|
||
|
||
- edge RecoveryBuffer + uplink 补传协议；
|
||
- RecoveryService：Gap 检测、冗余 binding 切换、历史回放；
|
||
- 与 Fusion/Subscription 联动 quality 状态机；
|
||
- 多 edge 负载与故障转移策略。
|
||
|
||
### C.5 其他基础设施
|
||
|
||
- D-Bus IPC 接入 CoreService；
|
||
- OPC UA 地址空间增量热重建；
|
||
- TLS/mTLS 生产部署；
|
||
- CommandExecutor 专用 egress 线程。
|
||
|
||
---
|
||
|
||
*文档版本：Phase 1 代码库 + 数据服务层演进规划（第 4 章）。术语定义见第 0.1 节。*
|