Separate the public docs (how to understand, deploy, and use IoT DC3) from engineering material (design drafts, analyses, roadmaps, legacy operation screenshots): - add an internal archive area with its own nav, grouping design, analysis, strategy, backlog, and legacy-operation sections - migrate design, operation (data/device/driver/template), analyses, and todo notes into the archive, preserving history - refresh public docs: README (en/zh/ja/vi), operation guides (agentic/concepts/data-commands/device-onboarding), architecture, development, guide, quickstart, and modules - promote the project overview to README.ai.md at the repo root
32 KiB
title
| title |
|---|
| 位号命令链路重构方案 |
位号命令链路重构方案
本文是 DC3 位号读写下行链路(HTTP/gRPC → 数据中心 → RabbitMQ → 驱动 → 设备)的重构与优化设计。
实施状态:方案已分阶段落地中。命名收敛 (Phase 1) 已全部完成;链路质量改造 (Phase 2-4) 部分完成,剩余项见各章节
[DONE]/[PARTIAL]/[TODO]标记。
0. 实施进度总览
| 章节 | 模块 | 状态 | 关键缺失 |
|---|---|---|---|
| §2 | 命名收敛 | [DONE] | — |
| §2.3 | 死代码清理 | [DONE] | — |
| §4.1 | DTO 序列化改造 | [DONE] | sealed interface + polymorphic payload + Instant 时间字段 + record DTO |
| §4.2 | 路由拓扑 (DLX + result) | [DONE] | — |
| §4.3 | dc3_point_command 持久化 |
[DONE] | — |
| §4.4 | 同步等待 API | [PARTIAL] | POST 返回 commandId, GET /{commandId} 可轮询; CompletableFuture 同步等待未实现 |
| §4.5 | Driver: write() 返回值 | [DONE] | Boolean 返回值被正确接收, 失败不 echo 假值 |
| §4.5 | Driver: 设备级串行锁 | [DONE] | DeviceLockManager.runExclusive(deviceId) 串行化同设备命令 |
| §4.6 | 幂等去重 | [DONE] | — |
| §4.7 | 中心侧 enableFlag / rwFlag 校验 | [DONE] | — |
| §4.7 | Driver 在线检查 | [DONE] | EntityStateMapper 查询 DRIVER 状态, 离线抛 ServiceException |
| §4.7 | 值范围校验 (PointCommandValidator) | [PARTIAL] | 基础非空校验已实现; min/max/enum/step 校验待 point_ext.constraints 落地 |
| §4.8 | publisher confirm → SENT | [DONE] | — |
| §4.5 | Driver: expireAt 预判 | [DONE] | PointCommandReceiver 入口处检查, 过期直接 EXPIRED |
| §4.4 | 查询/列表 API | [DONE] | GET /{commandId} + POST /list 分页过滤已实现 |
| §4.7 | commandId 幂等 | [DONE] | 调用方可选传 commandId, 重复提交返回已有状态 |
1. 背景
1.1 命名收敛(已完成)
原代码中同时存在三组类,经 Phase 1 已统一为 PointCommand*:
| 组别 | 模块 | 处置 |
|---|---|---|
PointValueCommand* |
dc3-common-data / dc3-common-facade | 已重命名为 PointCommand* |
DeviceCommand* |
dc3-common-model / dc3-common-driver | 已合并到 PointCommand* |
DriverCommand* (DriverCommandDTO / DriverCommandTypeEnum) |
dc3-common-model / dc3-common-constant | 已删除(死代码) |
RabbitMQ 资源已统一为 dc3.e.point_command / dc3.q.point_command.<svc> / dc3.r.point_command.<svc>。
1.2 与物模型 Command(设备服务)的语义边界
参见 物模型设计方案。物模型规划中 Command 表示读写 Point 之外的动作型服务
(重启、校准、切换模式等),定义在 Profile 下、由 Device 调用。
本方案的位号读写不是物模型层概念,而是属性维度的运行态访问能力。两者通过命名前缀分层:
| 概念 | 命名前缀 | 队列 / 表前缀 |
|---|---|---|
| 位号读写命令(本方案) | PointCommand |
dc3.e.point_command / dc3_point_command |
| 物模型设备服务(未来) | Command |
dc3.e.command / dc3_command |
1.3 当前下行链路问题(及已解决项)
| 维度 | 问题 | 状态 |
|---|---|---|
| 唯一性 | 报文无 ID;同 device 多线程消费 | [DONE] — commandId + 幂等去重 + DeviceLockManager 设备级串行锁 |
| 准确性 | 不校验 enableFlag / rwFlag / driver 在线 / 值范围; 离线 driver 命令 30s 后静默丢弃 | [PARTIAL] — enableFlag / rwFlag / driver 在线校验已实现; 值范围仅基础非空 |
| 结果反馈 | HTTP 立即 R.ok(); driver SDK 忽略 write() 返回值并 echo 假位号值 |
[DONE] — write() 返回值被正确接收, POST 返回 commandId 可轮询 |
| 序列化 | DTO 字段 content 是二次 JSON 序列化的 String; 缺 tenantId / commandId / source / expireAt |
[DONE] — sealed interface + polymorphic payload + record DTO |
2. 命名收敛 [DONE]
Phase 1 已全部完成。以下映射表保留作为历史参照。
2.1 重命名映射表
| 类别 | 旧名 | 新名 |
|---|---|---|
| HTTP 路径前缀 | /point_value_command |
/point_command |
| HTTP Controller | PointValueCommandController |
PointCommandController |
| Service 接口 | PointValueCommandService |
PointCommandService |
| Service 实现 | PointValueCommandServiceImpl |
PointCommandServiceImpl |
| 顶层 DTO | DeviceCommandDTO |
PointCommandDTO |
| 子结构(读) | DeviceCommandDTO.DeviceRead |
PointCommandDTO.PointRead |
| 子结构(写) | DeviceCommandDTO.DeviceWrite |
PointCommandDTO.PointWrite |
| 类型枚举 | DeviceCommandTypeEnum |
PointCommandTypeEnum |
| Exchange | dc3.e.command |
dc3.e.point_command |
| 命令 routing | dc3.r.command.device.<svc> |
dc3.r.point_command.<svc> |
| 命令队列 | dc3.q.command.device.<svc> |
dc3.q.point_command.<svc> |
| 死信 Exchange | — | dc3.e.point_command_dead |
| 死信队列 | — | dc3.q.point_command_dead |
| 结果 Exchange | — | dc3.e.point_command_result |
| 结果队列 | — | dc3.q.point_command_result |
| 常量类 | RabbitConstant.*COMMAND* |
RabbitConstant.*POINT_COMMAND* |
| Driver 监听器 | DeviceCommandReceiver |
PointCommandReceiver |
| Facade 接口 | PointValueCommandFacade |
PointCommandFacade |
| Facade 实现 | PointValueCommandGrpcFacade / PointValueCommandLocalFacade |
PointCommandGrpcFacade / PointCommandLocalFacade |
| 数据库表 | — | dc3_point_command |
2.2 死代码清理 [DONE]
已物理删除:
dc3-common-model/.../dto/DriverCommandDTO.javadc3-common-constant/.../enums/DriverCommandTypeEnum.java
3. 链路总体目标
┌──────────────┐ HTTP/JSON ┌────────────────────────┐ 持久化
│ 前端 / 第三方 │──────────▶│ dc3-center-data │──────▶ dc3_point_command
└──────────────┘ │ PointCommandController │ (PENDING)
│ PointCommandService │
┌──────────────┐ gRPC │ │
│ dc3-agentic │──────────▶ │ │
└──────────────┘ └────────────────────────┘
│
发布到 dc3.e.point_command (publisher confirm + mandatory)
│
▼
dc3.q.point_command.<svc> (durable, x-dead-letter-exchange)
│
▼
┌─────────────────────────────────┐
│ Driver: PointCommandReceiver │
│ - 幂等去重 (commandId) │
│ - 设备级串行 (deviceId 锁) │
│ - 调用 DriverCustomService.read/write │
└─────────────────────────────────┘
│
发布到 dc3.e.point_command_result
│
▼
dc3.q.point_command_result (中心侧消费)
│
▼
更新 dc3_point_command 状态
+ 推送 WebSocket / 同步释放挂起的 HTTP 请求
四个核心改动及实施状态:
- 报文层 [DONE]:引入
commandId、tenantId、source、occurredAt、expireAt、schemaVersion; 去掉content双重 JSON → sealed interface + polymorphic payload。 - 路由层 [DONE]:结果回执通道 + DLX 均已配置并运行。
- 存储层 [DONE]:
dc3_point_command表 +PointCommandDO+ 生命周期状态机已落地。 - 行为层 [DONE]:幂等去重 +
write()返回值 + 设备级串行锁 +expireAt预判均已实现。
4. 详细设计
4.1 命令报文(DTO)[DONE]
当前实现仍使用旧模型:
PointCommandDTO为@Builder类,content字段存放二次 JSON 序列化的PointRead/PointWrite字符串,时间字段为LocalDateTime,且implements Serializable。以下为设计目标。
完整 IDL 在落地 PR 中给出。
public sealed interface PointCommandPayload
permits PointCommandPayload.Read, PointCommandPayload.Write {
record Read(Long deviceId, Long pointId) implements PointCommandPayload {
}
record Write(Long deviceId, Long pointId, String value) implements PointCommandPayload {
}
}
public record PointCommandDTO(
String commandId, // UUID v4,调用方生成或中心侧补
Long tenantId, // 必填:跨租户隔离
PointCommandTypeEnum type, // READ / WRITE
PointCommandPayload payload, // 直接强类型,不再嵌套 JSON 字符串
CommandSource source, // HTTP / GRPC / AGENTIC / SCHEDULED
Long sourceUserId, // 发起人(HTTP 来源),可空
Instant occurredAt, // 调用方时间戳(UTC)
Instant expireAt, // 软超时,broker TTL 之外的应用层超时
int schemaVersion // 协议版本,初版 = 1
) {
}
要点:
payload用 sealed interface + Jackson polymorphic(@JsonTypeInfo(use = NAME, property = "kind")), 反序列化时按kind字段直接产出Read/Write,取消content字段的二次 JSON 序列化。- 所有时间字段使用
Instant(UTC),统一通过项目级ObjectMapper+JavaTimeModule配置;不接受裸LocalDateTime。 commandId强制通过 Jakarta Validation 校验为 36 位 UUID 字符串;调用方未填写则中心侧补 UUID v4。expireAt默认 =occurredAt + 10s,前端可在 VO 里覆盖;driver 收到时若已超期直接丢弃并发 TIMEOUT 结果。- 不再
implements Serializable,改为纯 record;DTO 仅经由 JSON 在 RabbitMQ 流转。
结果回执 DTO:
public record PointCommandResultDTO(
String commandId,
Long tenantId,
PointCommandStatusEnum status, // SUCCESS / FAILED / TIMEOUT / EXPIRED / DUPLICATE
String responseValue, // 成功时回写值,失败时为空
String errorCode, // 失败码,例如 PROTOCOL_REJECT / DEVICE_OFFLINE / VALIDATION
String errorMessage,
Instant finishedAt,
int schemaVersion
) {
}
4.2 路由拓扑 [DONE]
路由已全部配置并运行。DriverTopicConfig 声明 driver 侧命令队列(30s TTL + DLX),DataTopicConfig 声明中心侧死信队列与结果消费队列。
| Exchange | Queue | Routing Key | TTL / DLX |
|---|---|---|---|
dc3.e.point_command (topic) |
dc3.q.point_command.<svc> |
dc3.r.point_command.<svc> |
TTL=30s,DLX=dc3.e.point_command_dead |
dc3.e.point_command_result (topic) |
dc3.q.point_command_result |
dc3.r.point_command_result |
TTL=60s,DLX=dc3.e.point_command_dead |
dc3.e.point_command_dead (topic) |
dc3.q.point_command_dead |
# |
无 TTL,需人工/定时任务清理 |
要点:
- 命令队列引入 DLX,覆盖现状(reject 后静默丢失)的最大风险;DLX 队列的消息携带
x-death头,便于审计与重放。 - Driver 端继续以
serviceName为路由键,不变;多副本 driver 共享同一队列,competing consumers 语义保持。 - 结果回执通道是新加的,driver SDK 必须发;中心侧消费者更新命令状态 + 触发 WebSocket 推送。
4.3 持久化:dc3_point_command [DONE]
表已通过 MyBatis Plus 落地(PointCommandDO,PointCommandManager)。中心侧写入 PENDING → SENT,
PointCommandResultReceiver 推进到终态,PointCommandDeadReceiver 处理死信。
CREATE TABLE dc3_point_command
(
id BIGSERIAL PRIMARY KEY,
command_id CHAR(36) NOT NULL,
tenant_id BIGINT NOT NULL,
type VARCHAR(8) NOT NULL, -- READ / WRITE
device_id BIGINT NOT NULL,
point_id BIGINT NOT NULL,
request_value VARCHAR(256), -- 仅 WRITE
response_value VARCHAR(256), -- 反馈值
status VARCHAR(16) NOT NULL, -- PENDING / SENT / SUCCESS / FAILED / TIMEOUT / EXPIRED / DEAD
error_code VARCHAR(64),
error_message VARCHAR(1024),
source VARCHAR(16) NOT NULL,
source_user_id BIGINT,
occurred_at TIMESTAMPTZ NOT NULL,
sent_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
expire_at TIMESTAMPTZ NOT NULL,
schema_version SMALLINT NOT NULL,
create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
update_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (tenant_id, command_id)
);
CREATE INDEX idx_point_command_lookup
ON dc3_point_command (tenant_id, device_id, point_id, occurred_at DESC);
CREATE INDEX idx_point_command_pending
ON dc3_point_command (status, expire_at) WHERE status IN ('PENDING','SENT');
- 持久化只在中心侧做;driver 端依赖回执 + 自身日志,不引入额外存储。
- 表保留按
tenant_id分区或定期归档(默认 7 天,与点值历史的留存策略一致)。
4.4 中心侧 API [DONE]
| 方法 | 状态 | 说明 |
|---|---|---|
POST /point_command/read |
[DONE] | 返回 R<String> (commandId), 调用方可轮询结果 |
POST /point_command/write |
[DONE] | 同上 |
POST /point_command/submit-async |
[TODO] | 未实现(低优先级,当前 fire-and-forget 可通过 list 查状态) |
GET /point_command/{commandId} |
[DONE] | RESTful 路径, 返回 PointCommandDO 含当前状态 |
POST /point_command/list |
[DONE] | 分页查询,支持 deviceId/pointId/status/type 过滤 |
同步等待策略(待实现):
- 中心侧维护内存级
Map<commandId, CompletableFuture<Result>>; - 收到 result 消息时 complete future;
- HTTP 等待窗口 =
min(expireAt - now, 5s); - 超时返回
408 + 当前状态; - 超时后 future 不再阻塞调用方,但状态机依然受 result 通道驱动。
同步等待仅在 dc3-center-data 单实例内有效。多实例部署需选型:
- 简易方案(推荐先行):HTTP 短轮询 / SSE,前端 1s 一次查
GET /point_command/{id},最长 10s。- 进阶方案:Redis pub/sub 广播 result,由持有 future 的实例 complete。
4.5 Driver SDK 改动 [DONE]
已实现项:PointCommandReceiver 替代了 DeviceCommandReceiver;幂等去重 + 结果回执通道可用;write() 返回值被正确接收;
expireAt 预判已实现。
待实现项(按优先级):
| 项 | 设计 | 当前缺陷 |
|---|---|---|
| write() 返回值 | 必须接收 driverCustomService.write(...) 的 boolean 返回值 |
[DONE] |
| 真实读值进 result | read() 的真实值放入 result,不单靠 PointValue 流 |
responseValue 始终为 null(read 是异步的) |
| 设备级串行锁 | deviceLockManager.runExclusive(deviceId, ...) |
[DONE] DeviceLockManager 已集成到 PointCommandReceiver 分发链路 |
| expireAt 预判 | Instant.now().isAfter(dto.expireAt()) → 直接 EXPIRED |
[DONE] |
| PointCommandValidator | 从 point_ext 读取 min/max/enum/step 校验 |
基础非空已实现, 扩展校验待 point_ext.constraints 落地 |
4.6 幂等去重缓存 [DONE]
已实现 CommandDedupCache(Caffeine, 5min TTL, 50k max, putIfAbsent),位置:
dc3-common-driver/.../cache/CommandDedupCache.java。Duplicate 命令发 DUPLICATE 回执。
4.7 验证规则前置 [DONE]
| 校验 | 状态 | 失败处理 |
|---|---|---|
device.enableFlag = ENABLE |
[DONE] | 已在 PointCommandServiceImpl.validateScope() |
point.enableFlag = ENABLE |
[DONE] | 同上 |
WRITE 时 point.rwFlag 包含 WRITE |
[DONE] | 已在 validateWriteScope() |
driver 已注册且在线 |
[DONE] | EntityStateMapper 查 DRIVER 状态, 离线抛 ServiceException |
| WRITE 值基础非空 | [DONE] | PointCommandValidator.validateWriteValue() |
WRITE 值落入 point_ext 的 min/max/enum/step |
[TODO] | 待 point_ext.constraints 字段落地 |
commandId 在表中已存在 |
[DONE] | 调用方可选传 commandId, 重复提交返回已有 commandId |
4.8 publisher 侧确认 [DONE]
已实现:RabbitTemplate#correlationData = commandId,ConfirmCallback ACK 推进 PENDING → SENT。
5. 时序示例
当前实际路径中
complete future和200 OK + value两步未实现,HTTP 在 publish 后立即返回R.ok()。以下为设计目标。
写命令同步等回执的成功路径:
前端 Center-Data Broker Driver 设备
│ POST /write │ │ │ │
│ ─────────────────▶ │ INSERT command (PENDING) │ │ │
│ │ publish (correlationId=id) │ │ │
│ │ ───────────────────────────▶ │ │ │
│ │ ◀── confirm ACK ──── status=SENT │ │
│ │ │ ─── deliver ───────▶ │ │
│ │ │ │ dedup OK │
│ │ │ │ deviceLock acquire │
│ │ │ │ writeRegister ────▶│
│ │ │ │ ◀── ok ────────────│
│ │ │ │ ack command │
│ │ │ ◀── publish result ──│ │
│ │ ◀── deliver result ── │ │ │
│ │ status=SUCCESS │ │ │
│ │ complete future │ │ │
│ ◀── 200 OK + value │ │ │ │
异常路径下的关键差异:
- driver 抛异常 →
nack(requeue=true)+ 同步发一条 FAILED result(status 会被覆盖回滚SENT); - 二次失败 →
reject(requeue=false)进 DLX;中心侧 result 消费器收到x-death后把状态推进到DEAD。
6. 升级说明
Phase 1(命名一刀切)已完成并发布。旧 HTTP 路径、旧 exchange/queue、旧 DTO 已删除。新 driver 与旧中心不能混跑。
6.1 剩余改造的升级策略
剩余 Phase 2-4 改造对以下模块的要求:
| 模块 | 影响 | 升级动作 |
|---|---|---|
| iot-dc3-web | HTTP API 语义变化(同步返回真实结果) | 切换到新 API 语义;新增"命令历史"页面 |
| dc3-center-agentic | gRPC stub 命名变化 | 重新生成 stub |
| dc3-driver-* | SDK 行为变化(write 返回值、设备锁、值校验) | 升级 SDK 依赖 |
6.2 数据库变更
dc3_point_command表已创建(DDL 见 §4.3);- 预留:
dc3_point.point_ext增加constraints字段(min / max / enum / step / readonly), 缺失时跳过值校验,不阻塞发版。
7. 落地阶段(修订)
原 Phase 1 已全部完成。Phase 2-4 按实际进度重新划分优先级:
| 优先级 | 步骤 | 范围 | 出口标准 |
|---|---|---|---|
| P0 | DTO 序列化改造 | sealed interface + polymorphic payload;Instant 时间字段;commandId / expireAt / source / schemaVersion |
[DONE] content JSON 字符串字段删除;DTO 为 record 类型 |
| P0 | write() 返回值 | DriverWriteServiceImpl 接收 Boolean 返回值;失败不 echo 假值 |
[DONE] 写失败时 result.status = FAILED, responseValue 为空 |
| P1 | 同步等待 / 轮询 API | POST /point_command/* 返回 commandId;GET /point_command/{id} 轮询 |
[PARTIAL] 轮询已可用; CompletableFuture 同步等待未实现 |
| P1 | Driver 在线 + 值范围校验 | 中心侧在线检查;PointCommandValidator |
[PARTIAL] 在线检查已实现; 值范围仅基础非空 |
| P2 | 设备级串行锁 | deviceLockManager.runExclusive(deviceId) |
[DONE] 同 device 并发写无协议交错 |
| P2 | expireAt 预判 | PointCommandReceiver 入口处检查 expireAt |
[DONE] 过期命令直接 EXPIRED,不进业务逻辑 |
| P3 | 查询/列表 API 完善 | GET /point_command/{commandId} 规范 URL;GET /point_command/list 检索 |
[DONE] RESTful 查询 + 分页列表已实现 |
| P3 | DLX 重放工具 | CLI 或后台页面消费 dc3.q.point_command_dead |
[TODO] 运维可从死信队列恢复命令 |
每个步骤可独立 PR,不要求同一 release。
8. 性能与容量预估
dc3_point_command写入 QPS ≈ HTTP 写入 QPS。在 1k QPS 下,按 7 天保留 ≈ 6 亿行, 需要在tenant_id, occurred_at上做范围分区或冷归档,不在初版要求范围内但要预留。- 同步等待用的
CompletableFuture数量 ≈ 并发 HTTP 请求数;超时强制释放,无溢出风险。 - driver 端
Caffeinededup 缓存:50_000 条 × 36B ≈ 2MB,可忽略。 - result 通道双倍 RabbitMQ 流量;按现有点值流量 1/100 估算,broker 压力增量极小。
9. 风险与未决项
| 项 | 状态 | 风险 | 应对 |
|---|---|---|---|
| 同步等待 vs 异步轮询 | 未决 | 多实例部署时同步等待需要广播 result | 简易方案:202 + 轮询;进阶方案:Redis pub/sub |
commandId 是否对前端必填 |
已决 | 前端忘填会丢失天然幂等 | 中心侧补 UUID,同时推荐前端预生成 |
物模型 Command 何时落地 |
未决 | 需要 dc3.e.command / dc3_command 命名空间 |
物模型方案 PR 中显式约束命名前缀 |
| DLX 重放工具 | 未决 | 无运维工具消费死信队列 | P3 单独立项 |
| Driver 进程级 dedup | 已知约束 | driver 重启丢去重窗口 | 属于设计取舍,生产可接受 |
| 命名切换对生产环境的冲击 | 已解决 | Phase 1 已发布,旧资源已删除 | — |
| sealed interface + record 兼容性 | 新风险 | Jackson 2.x 对 Java record 的 parameter name 支持 | 落地前验证 ParameterNamesModule;不通过则降级为普通类 + @JsonTypeInfo |
10. 决策项
- 命名
PointCommand— 已确认,Phase 1 已完成。 - 同步等回执范围 — 待决:P1 优先采用简易方案(202 + 轮询),多实例广播延后。
- DLX 重放工具 — 待决:P3 单独立项。
- WRITE 值范围校验位置 — 已确认:中心侧(统一)+ driver 侧(贴协议)两层兜底。
schema_version— 已确认:进库,dc3_point_command表已包含schema_version SMALLINT列。
11. 引用
- 物模型设计:docs/superpowers/design/thing-model.md
- 当前代码定位(Phase 1 后的命名):
dc3-common-data/.../controller/PointCommandController.javadc3-common-data/.../biz/impl/PointCommandServiceImpl.javadc3-common-data/.../receiver/rabbit/PointCommandResultReceiver.javadc3-common-data/.../receiver/rabbit/PointCommandDeadReceiver.javadc3-common-model/.../dto/PointCommandDTO.javadc3-common-model/.../dto/PointCommandResultDTO.javadc3-common-driver/.../receiver/rabbit/PointCommandReceiver.javadc3-common-driver/.../service/impl/DriverWriteServiceImpl.javadc3-common-driver/.../service/impl/DriverReadServiceImpl.javadc3-common-driver/.../cache/CommandDedupCache.javadc3-common-driver/.../config/DriverTopicConfig.javadc3-common-rabbitmq/.../config/RabbitConfig.javadc3-common-constant/.../enums/PointCommandTypeEnum.javadc3-common-constant/.../enums/PointCommandStatusEnum.javadc3-common-constant/.../enums/PointCommandSourceEnum.javadc3-common-data/.../entity/model/PointCommandDO.java