# 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 节。*