分布式方案设计

#技术

设备巡检主节点 Java 后台架构 — 正解设计稿（v2.0）

替代第一版评审稿。本稿目标是把"承诺了语义、没承诺实现"的所有空洞都补齐，给出一套生产可落地、能闭环、能承受 5x 流量压测的完整设计。

0. 执行摘要 — 九个核心决策

1. 原设计根因分析 — 为什么会陷入混乱

你提的核心痛点是："版本机制有两套、排水时 IS 不知道该不该投递、扩缩容导致全量对齐"。这背后是三个本来应该独立的概念被强行用同一个"槽点版本"表达：

2. 模型层 — 五个版本号、三个状态机

2.1 五个版本号的精确定义

2.2 三个状态机

Worker 状态机： [*] → REGISTERED → WARMING → ACTIVE ↓ ┌─────────── DRAINING ─→ OFFLINE │ ↑ └─→ SUSPECTED → DEAD → QUARANTINE → WARMING (心跳超) (隔离观察)

                 │
                 └─→ ORPHANED → LOADING_NEW → ACTIVE
                      (旧主死)

                                          ↓ fencing 失败
                                     ABORT_BATCH → STOP_CONSUMER

2.3 Slot Lease 表结构（DDL）

CREATE TABLE slot_lease ( slot_id SMALLINT PRIMARY KEY, owner_worker_id VARCHAR(64), -- NULL 表示无主 lease_epoch BIGINT NOT NULL DEFAULT 0, -- 单调递增，永不回退 state VARCHAR(16) NOT NULL, -- ACTIVE / MOVING / DRAINING_OLD / LOADING_NEW / ORPHANED / FREE lease_until TIMESTAMPTZ, -- 当前租约到期时刻 last_renewed_at TIMESTAMPTZ, slot_map_version BIGINT NOT NULL, -- 该 slot 当前生效的 device 映射版本 -- 用于 grace window 期间识别 device 跨 slot weight INT NOT NULL DEFAULT 1 );

CREATE INDEX idx_slot_lease_owner ON slot_lease(owner_worker_id); CREATE INDEX idx_slot_lease_state ON slot_lease(state); lease_epoch 是核心 fencing 字段：每次 owner 切换 +1，永不回退。Web 调度器是唯一的 epoch 写入方，Worker 只读。

2.4 消息信封格式（IS → MQ → Worker）

{ "schema_version": 2, "envelope": { "message_id": "<is_id>:<cycle_id>:<device_id>:", "device_id": "dev_00012345", "slot_id": 73, "slot_map_version": 1024, "resource_version": 17, "plan_version": 9, "is_node": "is-03", "cycle_id": 20260425103000, "seq": 42, "collected_at": "2026-04-25T10:30:01.123Z", "server_received_at": null, "priority": "high" }, "payload": { /* 巡检状态字段 */ } }

3. 实现层 — 把语义落到代码

3.1 Worker Fencing 的具体实现（解决 D3）

核心思想：每次 Worker 处理一批消息时，写入 device_status_current 必须带两个谓词：

1. incoming_seq > current.seq（防旧覆盖新）
2. incoming_writer_epoch == current_lease_epoch（防旧 owner 写入）

⠀不依赖事务级 SELECT FOR UPDATE，因为那会让 slot 切换被 Worker 长事务阻塞。 device_status_current 表关键列： CREATE TABLE device_status_current ( device_id VARCHAR(64) PRIMARY KEY, slot_id SMALLINT NOT NULL, current_slot SMALLINT NOT NULL, -- grace window 期间区分新旧 slot seq BIGINT NOT NULL, cycle_id BIGINT NOT NULL, collected_at TIMESTAMPTZ NOT NULL, writer_epoch BIGINT NOT NULL, -- 上次写入者持有的 lease_epoch status_payload JSONB NOT NULL, derived_snapshot JSONB, -- 派生状态快照（防抖计数、滑动窗口、上次心跳等） updated_at TIMESTAMPTZ NOT NULL DEFAULT now() );

CREATE INDEX idx_dsc_slot ON device_status_current(slot_id); CREATE INDEX idx_dsc_updated ON device_status_current(updated_at); Fencing UPSERT 模板（这是 Worker 必须使用的固定模板，做进 SDK）： -- 单条 upsert 模板（批量执行时用 UNNEST 多行同时写） INSERT INTO device_status_current (device_id, slot_id, current_slot, seq, cycle_id, collected_at, writer_epoch, status_payload, derived_snapshot, updated_at) VALUES ($1,$2, $2,$3, $4,$5, $6,$7, $8, now()) ON CONFLICT (device_id) DO UPDATE SET seq = EXCLUDED.seq, cycle_id = EXCLUDED.cycle_id, collected_at = EXCLUDED.collected_at, writer_epoch = EXCLUDED.writer_epoch, current_slot = EXCLUDED.current_slot, status_payload = EXCLUDED.status_payload, derived_snapshot = EXCLUDED.derived_snapshot, updated_at = now() WHERE -- 旧消息防御 device_status_current.seq < EXCLUDED.seq -- Fencing：旧 owner 写不进来（epoch 已升） AND EXCLUDED.writer_epoch >= device_status_current.writer_epoch -- Grace window 期间：只有当前合法 slot 的 owner 才能写 AND (device_status_current.current_slot = EXCLUDED.slot_id OR device_status_current.current_slot IS NULL); 为什么不用事务级 fencing：

• 事务级 SELECT lease_epoch FOR UPDATE 会持有锁直到 commit，长事务时 Web 调度器无法切换该 slot
• 行级 fencing 的代价是每行 UPSERT 多两个谓词比较（B-Tree 索引上常数开销），可接受
• 副作用：writer_epoch >= current.writer_epoch 允许同 epoch 写入，配合 seq < EXCLUDED.seq 仍然防住旧消息

⠀Worker 批量提交伪代码： function commitBatch(slotId, messages): # 1. 应用侧去重：同 device 取 seq 最大的那条 deduped = dedupByDeviceLatestSeq(messages)

# 2. 读当前持有的 lease epoch（来自 Worker 内存，由心跳线程刷新）
myEpoch = currentLeases[slotId].epoch
if myEpoch is null:
    abortBatch(messages)  # 不 ack，等重投递
    return

# 3. 批量 UPSERT current（行级 fencing）
BEGIN
    UNNEST 所有 deduped 一次性发出
    rowsAffected = 实际更新的行数
    
    # 4. 批量 INSERT history（append-only，不需要 fencing）
    INSERT INTO device_status_history (...) VALUES ...
    
    # 5. 批量 INSERT alert_event（append-only）
    INSERT INTO alert_event (...) VALUES ...
    
    # 6. 批量更新 alert_current（基于 alert_event 物化）—— 见 §3.3
    merge_alert_current_from_events(...)
    
    # 7. 批量更新 rollup（基于 EXCLUDED 增量）
    upsert_rollup(...)
COMMIT

# 8. 提交成功才 ack
if commit success:
    for msg in messages:
        channel.basicAck(msg.deliveryTag, multiple=false)
else:
    # 不 ack，等待 RabbitMQ 重投递
    log error
    # 如果是 fencing 失败（不是 DB 故障），主动 cancel 该 slot 的 consumer
    if fencingFailed:
        cancelSlotConsumer(slotId)

关键点：

• rowsAffected 小于 deduped.size 是正常的（fencing 拒绝了旧消息），但仍然 ack——这些消息已经被合法处理（被新消息替代）。
• fencing 失败（比如 epoch 已升）说明本 Worker 不再是 owner，应当主动 cancel consumer，让消息回流给新 owner。
• 不要重试 fencing 失败的事务，会产生死循环。

⠀3.2 派生状态持久化（解决 D5） derived_snapshot JSONB 列存放 SlotProcessor 内的所有派生状态。设计原则： 原则：device 当前状态 = fold(历史所有事件) = fold(快照截至 T0 的状态, T0 之后的事件) 典型派生字段： { "consec_failure_count": 3, "last_n_states_ring": ["UP","UP","DOWN","UP","DOWN"], "ring_pos": 4, "last_heartbeat_at": "2026-04-25T10:29:50Z", "rate_window": { "window_start": "2026-04-25T10:25:00Z", "events_count": 142, "last_event_at": "2026-04-25T10:29:58Z" }, "debounce_pending": { "since": "2026-04-25T10:29:30Z", "trigger_threshold": 5 } } 约束： INV-8: SlotProcessor 内不允许有任何"未持久化的状态"。 INV-9: 每次 batch commit 必须把派生状态作为快照写入 derived_snapshot。 INV-10: Worker 接管 slot 时，加载所有受影响 device 的 derived_snapshot 到内存。 Slot 迁移加载流程： 1. 新 Worker 收到 slot 分配 2. SELECT device_id, status_payload, derived_snapshot, seq, cycle_id FROM device_status_current WHERE current_slot = ? 分批（如每批 1000 行）加载到 SlotProcessor 内存 3. 加载完成后，Worker 上报 ready 4. Web 把 slot 状态从 LOADING_NEW 切到 ACTIVE 5. Worker 开始 basic.consume 容量估算：单 device snapshot ≤ 2KB，1M device 总快照量约 2GB，单 slot（4000 device）约 8MB，加载时间 < 1 秒（DB 不忙时）。

3.3 告警语义重构（解决 D4）

原设计的"newest seq wins" 不能用于告警，因为： 高优先级告警 firing (cycle=10, seq=100) 投到 high queue 普通状态更新 status (cycle=11, seq=101) 投到 normal queue Worker 处理顺序：先 status，再 firing "newest seq wins" → status (seq=101) 覆盖 firing (seq=100) → 告警丢失 修正：告警事件流为事实源，alert_current 由事件流确定性派生。 alert_event 表（append-only，事实源）： CREATE TABLE alert_event ( event_id BIGSERIAL PRIMARY KEY, device_id VARCHAR(64) NOT NULL, alert_key VARCHAR(128) NOT NULL, -- 同一类告警的稳定 key（如 "device.offline"） event_type VARCHAR(16) NOT NULL, -- FIRING / RESOLVED / UPDATED severity SMALLINT NOT NULL, -- 0-4: INFO/WARN/MINOR/MAJOR/CRITICAL event_time TIMESTAMPTZ NOT NULL, -- 事件发生时刻（来自 collected_at） cycle_id BIGINT NOT NULL, seq BIGINT NOT NULL, payload JSONB NOT NULL, ingest_writer_epoch BIGINT NOT NULL, UNIQUE (device_id, alert_key, cycle_id, seq) );

CREATE INDEX idx_ae_device_key_time ON alert_event(device_id, alert_key, event_time DESC); UNIQUE 约束直接做幂等：重复消息 INSERT ON CONFLICT DO NOTHING 即可。 alert_current 表（物化视图，最后一次有效告警状态）： CREATE TABLE alert_current ( device_id VARCHAR(64) NOT NULL, alert_key VARCHAR(128) NOT NULL, state VARCHAR(16) NOT NULL, -- ACTIVE / RESOLVED severity SMALLINT NOT NULL, first_fired_at TIMESTAMPTZ NOT NULL, last_event_at TIMESTAMPTZ NOT NULL, last_event_id BIGINT NOT NULL, -- 引用 alert_event.event_id payload JSONB NOT NULL, PRIMARY KEY (device_id, alert_key) ); 事件 → 当前态的派生规则： 对于同一 (device_id, alert_key)，按 event_time 排序，规约：

• 最近的 FIRING/UPDATED 之后没有 RESOLVED → ACTIVE

• 最近的事件是 RESOLVED → RESOLVED

• severity 取最近 ACTIVE 期间的最大值 Worker 处理告警的伪代码： function processAlertEvent(message, currentDeviceState):

  1. INSERT alert_event（带 UNIQUE 约束，重复自动忽略）

  INSERT INTO alert_event (...) ON CONFLICT DO NOTHING

  2. 重新计算 alert_current（仅查询近期事件，不扫全表）

  latestActive = SELECT MAX(event_time), MAX(severity) FROM alert_event WHERE device_id=? AND alert_key=? AND event_time > now() - interval '7 days' AND event_id > (SELECT COALESCE(last_event_id,0) FROM alert_current ...)

  3. UPSERT alert_current（fencing 同样适用）

  UPSERT alert_current ... 性能保护：alert_event 按 (device_id, alert_key) hash 分区或按 event_time range 分区，查询永远命中索引，不退化。

3.4 Grace Window 双写隔离（解决 D3 + 重分片场景）

device → slot 真正变更（slot_map_version 升）时，宽限期内同 device 可能存在新旧两个 slot 各自的 Worker 同时写入。 协议： T0: Web 发布 slot_map_version=N+1，effective_time=T0+5min grace window = [T0, T0+10min]

T0~T0+5min: IS 仍按 v=N 投递，新 IS 拉到 v=N+1 但 effective_time 未到， 依然按 v=N 投递。Worker 接受 v=N 消息正常。

T0+5min: IS 切到 v=N+1。开始按新 slot 投递。 Worker 进入"双版本接受"模式：v=N 和 v=N+1 都接收。

T0+5min ~ T0+10min: 对受影响 device，新旧两个 slot 都可能收到消息。 Web 在 effective_time 这一刻，对每个受影响 device 执行： UPDATE device_status_current SET current_slot = newSlot, slot_id = newSlot WHERE device_id = ?

          旧 slot 的 Worker 此后 UPSERT 会因 current_slot != EXCLUDED.slot_id 失败 → 不更新。
          新 slot 的 Worker UPSERT 会成功。
          旧 slot 的 history 仍然 INSERT（append-only，无害）。

T0+10min: Worker 拒绝任何 v=N 的消息（已超出 grace window）。 IS 端若仍有 v=N 的本地 buffer 消息，按设备级 message_id 幂等忽略。 关键 fencing 谓词（已包含在 §3.1 的 SQL 模板里）： AND (device_status_current.current_slot = EXCLUDED.slot_id OR device_status_current.current_slot IS NULL); 重分片操作清单（一次性 DDL，由 Web 后台执行）： BEGIN; -- 1. 标记新 slot_map_version INSERT INTO slot_map_version (version, effective_at) VALUES (N+1, '...');

-- 2. 计算受影响设备 INSERT INTO slot_remap_pending (device_id, old_slot, new_slot) SELECT device_id, old_slot_fn(device_id), new_slot_fn(device_id) FROM device_resource WHERE old_slot_fn(device_id) <> new_slot_fn(device_id);

-- 3. 在 effective_time 时点，原子地切换 current_slot -- （由后台调度器在 effective_at 时刻执行） UPDATE device_status_current dsc SET current_slot = srp.new_slot, slot_id = srp.new_slot FROM slot_remap_pending srp WHERE dsc.device_id = srp.device_id;

-- 4. 通知 IS 切换 slot_map_version COMMIT; 评审建议：日常运维永远不要触发 slot_map_version 变更。这条流程只在以下场景使用：

• TOTAL_SLOTS 不够用，必须扩展
• 单 slot 长期热点无法用 owner 迁移解决
• 业务重大重组，分组规则改变

⠀3.5 IS WAL Buffer（解决 D6） 每个 IS 节点本地维护一个 SQLite/RocksDB WAL，所有产出消息先落盘再投 MQ。 SQLite 表： CREATE TABLE outbox ( seq INTEGER PRIMARY KEY AUTOINCREMENT, message_id TEXT NOT NULL UNIQUE, routing_key TEXT NOT NULL, body BLOB NOT NULL, created_at INTEGER NOT NULL, confirmed INTEGER NOT NULL DEFAULT 0 -- 0=未确认, 1=已 confirm );

CREATE INDEX idx_outbox_unconfirmed ON outbox(confirmed, seq) WHERE confirmed = 0; 生产侧伪代码： function produce(message): INSERT INTO outbox (message_id, routing_key, body, created_at) VALUES (...) sqlite.commit() # 落盘

# 异步投递
rabbitChannel.publish(routing_key, body, mandatory=true, with_confirm)

function onPublishConfirm(deliveryTag): UPDATE outbox SET confirmed=1 WHERE seq=mapping[deliveryTag]

function onPublishNack(deliveryTag): # 不更新 confirmed=1，重发任务会重新投递

function periodicReplayer(): # 每 5 秒一次 for row in SELECT * FROM outbox WHERE confirmed=0 AND created_at < now()-3s: rabbitChannel.publish(row.routing_key, row.body)

function periodicCleaner(): # 每 1 分钟一次 DELETE FROM outbox WHERE confirmed=1 AND created_at < now() - retention_period 容量规划：

1. 打开 SQLite，检查未 confirmed 记录
2. 按 seq 升序重放未 confirmed 的消息（保证设备级顺序）
3. 重放完成后，开始新一轮巡检
4. Worker 端依赖 message_id UNIQUE 约束去重 关键不变量： INV-11: IS 不允许"先发 MQ 后写 outbox"，必须先持久化。 INV-12: outbox 的 message_id 必须能在 IS 重启后被重新构造（基于 device_id + cycle_id + seq）。 INV-13: outbox retention ≥ MQ 消息 TTL + 安全余量，避免 dedup 表先于 outbox 过期。 SQLite 选型理由：单文件、ACID、libsqlite 直接嵌入 C++ 探针进程，无运维成本。RocksDB 适合更高吞吐场景但运维更重；当前 60 msg/s 的写入量，SQLite WAL 模式下完全胜任（实测可承受 10K writes/s）。

3.6 幂等键定义

message_id = "{is_node}:{cycle_id}:{device_id}:{seq}" message_dedup 表： CREATE TABLE message_dedup ( message_id VARCHAR(128) PRIMARY KEY, processed_at TIMESTAMPTZ NOT NULL DEFAULT now() ) PARTITION BY RANGE (processed_at);

-- 按天分区，保留 7 天 CREATE TABLE message_dedup_20260425 PARTITION OF message_dedup FOR VALUES FROM ('2026-04-25') TO ('2026-04-26'); Worker 处理时： 1. 批量 INSERT INTO message_dedup ... ON CONFLICT DO NOTHING 2. RETURNING 子句返回真正插入的 message_id 集合 3. 只对真正插入的消息执行业务处理 4. 无论是否首次见到，都 ack（重复消息已经被忽略） 保留策略：分区按天，保留 7 天即可（远长于 RabbitMQ 重投递窗口和 IS outbox retention）。

4. 调度协议 — Web 控制面的精确行为

4.1 Web 控制面 HA（解决 D7）

多实例 + Leader 选主： -- 在 PostgreSQL 上用 advisory lock 选主 SELECT pg_try_advisory_lock(hashtext('web_scheduler_leader')); -- 返回 true 即为 leader，定期续期（每 5 秒） 只有 leader 写入 slot_lease、做 rebalance 决策。Follower 只对外提供只读 API（前端查询、IS 拉资源）。 通知机制： 1. Leader 决策完成 → UPDATE slot_lease SET ... → COMMIT 2. Leader → Redis publish "slot_lease_updated" (slot_id 列表) 3. Worker 订阅 Redis → 收到通知 → 重新查询自己关心的 slot_lease 行 ↓ 兜底：Worker 每 5 秒主动 SELECT 一次自己持有的所有 lease，刷新本地缓存 4. Web Follower 也订阅 Redis → 刷新 dashboard 数据 关键不变量： INV-14: DB 中的 slot_lease 是事实源。Redis 仅为加速通知，丢消息不影响正确性。 INV-15: Worker 每次 batch commit 前必须确认本地 lease 缓存年龄 < 5 秒。 INV-16: Worker 收到 MQ delivery 时，若 slot 不在本地 lease 缓存中或 lease 已失效， 立即 basicCancel 该 channel 并 nack（requeue=true）。

4.2 Worker 心跳与租约续期

CREATE TABLE worker_heartbeat ( worker_id VARCHAR(64) PRIMARY KEY, state VARCHAR(16) NOT NULL, last_heartbeat TIMESTAMPTZ NOT NULL, metadata JSONB, -- 负载、版本、IP 等 backlog_metrics JSONB -- 积压、lag、处理延迟 ); Worker 心跳逻辑： 每 3 秒：

1. UPDATE worker_heartbeat SET last_heartbeat=now(), backlog_metrics=... WHERE worker_id=? RETURNING state
2. 如果 state=DEAD/QUARANTINE，立即停止所有消费
3. UPDATE slot_lease SET last_renewed_at=now() WHERE owner_worker_id=? AND lease_epoch=本地缓存的 epoch
4. 若 UPDATE 影响行数 < 本地持有 slot 数，说明部分 slot 被剥离 → 同步本地缓存 Web 调度器检测异常 Worker： 每 10 秒： WorkerSet = SELECT * FROM worker_heartbeat WHERE state=ACTIVE 对每个 worker: if now() - last_heartbeat > 15s: 标记为 SUSPECTED if now() - last_heartbeat > 30s: 标记为 DEAD，剥离全部 slot if state=DEAD AND now() - last_heartbeat > 90s: 清理记录 lease 超时强制接管： 每 10 秒： ExpiredSlots = SELECT slot_id FROM slot_lease WHERE state=ACTIVE AND lease_until < now() - 5s for slot in ExpiredSlots: UPDATE slot_lease SET owner_worker_id=NULL, state=ORPHANED, lease_epoch=lease_epoch+1 WHERE slot_id=? AND lease_epoch=旧值 -- 然后选一个 ACTIVE Worker 接管

4.3 Slot 调度算法（rebalance）

触发时机：

• 新 Worker 进入 ACTIVE
• Worker 进入 DRAINING/DEAD
• 周期性检查（每 60 秒）发现负载不均衡

⠀slot 权重计算： weight(slot) = α × device_count(slot) # 设备数 + β × msg_rate(slot, last_5min) # 消息速率 + γ × backlog_depth(slot) # MQ 积压 + δ × avg_process_ms(slot) # 处理耗时 + ε × high_priority_device_count(slot) # 高优先级设备数

默认参数

α = 1, β = 5, γ = 10, δ = 2, ε = 3 rebalance 决策： function rebalance(): workers = ACTIVE Worker 列表（state=ACTIVE 且 WARMING 已过） if len(workers) == 0: return

totalWeight = sum(weight of all ACTIVE slots)
targetPerWorker = totalWeight / len(workers)

# 找出最重和最轻的 Worker
loads = {w: sum(weight(s) for s in w.slots) for w in workers}
maxW = max(loads), minW = min(loads)

# 阈值：差距 > 25% 才触发
if (loads[maxW] - loads[minW]) / targetPerWorker < 0.25:
    return

# 单次最多迁移 N 个 slot（避免抖动）
movePlan = []
for slot in maxW.slots ordered by weight desc:
    if loads[maxW] - weight(slot) >= targetPerWorker - 5%:
        movePlan.append((slot, maxW, minW))
        if len(movePlan) >= 5: break

for (slot, from, to) in movePlan:
    executeMigration(slot, from, to)

单 slot 迁移流程（与 §4.4 中 graceful drain 一致）： function executeMigration(slot, oldOwner, newOwner): # 1. 标记 MOVING UPDATE slot_lease SET state=MOVING WHERE slot_id=?

# 2. 通知旧 owner 停止接收新消息
redis.publish("slot_command", {type: "DRAIN", slot: slot})

# 3. 旧 owner 收到命令：
#    - basicCancel 该 slot 的 consumer
#    - 处理本地 mailbox 内的消息
#    - commit 完毕后 → UPDATE slot_lease SET state=DRAINING_OLD
#    - 等待 ack 全部完成 → 通知 Web 释放

# 4. Web 收到 release 信号:
UPDATE slot_lease 
SET owner_worker_id=newOwner, lease_epoch=lease_epoch+1, 
    state=LOADING_NEW, lease_until=now()+30s
WHERE slot_id=?

# 5. 通知新 owner
redis.publish("slot_command", {type: "ASSIGN", slot: slot, epoch: newEpoch})

# 6. 新 owner 加载 derived_snapshot → 上报 ready
UPDATE slot_lease SET state=ACTIVE WHERE slot_id=?

# 7. 新 owner basic.consume

冷却约束：

• 单次 rebalance 最多迁 5 个 slot
• 同一 slot 两次迁移间隔 ≥ 5 分钟
• 同一 Worker 两次迁出间隔 ≥ 1 分钟

4.4 七种生命周期场景的完整闭环

场景 A：冷启动（系统首次安装） 1. DBA 初始化 TOTAL_SLOTS=256，slot_lease 全部 owner=NULL, state=FREE 2. Web 启动，竞选 leader 3. 第一个 Worker 启动 → 注册 → WARMING 4. WARMING 超期（60s 心跳稳定）→ ACTIVE 5. Web rebalance → 把所有 256 slot 分配给唯一的 Worker 注意：Worker 可能撑不住全部，按"线性逐步开启"策略 6. IS 启动 → 拉 resource_version、plan_version、slot_map_version 7. IS 按"渐进开闸"策略：先以 1/4 巡检频率运行，分批升频，避免新 Worker 被冲垮 8. 第二个 Worker 加入 → ACTIVE → rebalance 把 128 slot 迁过去 9. 第三个、第四个 Worker 陆续加入，slot 分布逐渐均匀 冷启动渐进开闸： 新 Worker 数 < 期望最小数（如 3）：

• IS 巡检频率降为正常的 25%
• 仅核心设备保持正常频率
• 每加入一个 Worker，全局频率升 25% 直到 Worker 数 ≥ 期望最小数 → 全速运行 场景 B：新 Worker 加入 1. Worker_New 启动 → 注册 → WARMING 2. 60 秒 WARMING 后 → ACTIVE 3. Web rebalance 触发 → 选 5 个 slot 从重负载 Worker 迁出 4. 每个 slot 走 §4.3 的 executeMigration 流程 5. 整体迁移 ≤ 1 分钟完成 6. IS 完全无感知 场景 C：Worker 优雅下线 1. 运维下发命令 → API 调用 /worker/{id}/drain 2. Web → UPDATE worker_heartbeat SET state=DRAINING 3. Web 不再给该 Worker 分配新 slot 4. Web 分批迁出该 Worker 持有的 slot（每批 5 个，间隔 30 秒） 5. 全部迁出后，UPDATE worker_heartbeat SET state=OFFLINE 6. Worker 自检发现 state=OFFLINE → 关闭所有连接 → 进程退出 场景 D：Worker 突然 kill -9 1. Worker 崩溃 → MQ TCP 连接断开 → 未 ack 消息回到队列 2. heartbeat 不再更新 → 15 秒后被标记 SUSPECTED 3. 30 秒后被标记 DEAD 4. Web 把该 Worker 的所有 slot 标记 ORPHANED 5. Web rebalance → 把这些 slot 分配给其它 ACTIVE Worker，lease_epoch+1 6. 新 owner 加载 derived_snapshot → 开始消费 7. 重投递的消息被新 owner 处理，message_dedup 防止重复 8. 所有 fencing 写入因 epoch 校验通过而成功 RTO 估算：心跳 15s + DEAD 判定 15s + 接管 5s + 加载快照 5s ≈ 40 秒。 场景 E：Worker 假死（关键场景） 1. Worker 心跳线程卡死，业务线程仍在消费 2. 30 秒后被标记 DEAD，slot 被剥离给新 Worker，lease_epoch+1 3. 旧 Worker 仍试图 commit batch 4. 在 commit 时：
• UPSERT 谓词 EXCLUDED.writer_epoch >= dsc.writer_epoch 失败 （新 Worker 已经写入了更高 epoch）
• 所有写入被拒绝
• rowsAffected = 0
• Worker 检测到 rowsAffected == 0 → 推断自己 lease 已失效
• 不 ack MQ，主动 cancel consumer，进入 SELF_QUARANTINE 5. Worker 内部告警 → 自杀重启 → 进入 WARMING 关键防御：fencing 在行级。即使 Worker 完全失控、绕过所有应用层检查，DB 也不会接受它的写入。 场景 F：DB 故障（停 5 分钟） 1. Worker UPSERT 失败 → 不 ack 2. RabbitMQ backlog 上升 3. Web 检测到全 Worker DB error → 通过 Redis 通知所有 IS：进入 BACKPRESSURE 模式 4. IS 收到通知：
• 核心设备：保持巡检，全部进 outbox
• 普通设备：降频 50%
• 低优先级：暂停巡检 5. 5 分钟后 DB 恢复，Web 通知 IS 解除 BACKPRESSURE 6. Worker 开始消费 backlog（prefetch 限流防止冲垮） 7. 1~2 分钟内 backlog 消耗完 场景 G：RabbitMQ 故障（停 5 分钟） 1. IS publish confirm 失败 2. IS outbox 中 unconfirmed 消息累积 3. 5 分钟内 SQLite outbox 增长 ~750MB / IS（按 60 msg/s） 4. RabbitMQ 恢复，IS replayer 按 seq 升序重发 5. Worker 端依赖 message_dedup 去重 6. 30 秒到 1 分钟内重发完成

4.5 重分片协议（slot_map_version 升级）

仅当以下场景触发：

• TOTAL_SLOTS 调整
• 单 slot 长期热点（> 7 天 weight 超均值 3 倍）
• 业务规则重组

⠀流程：见 §3.4。

5. MQ 与队列设计

5.1 队列拓扑（最终方案）

优先级简化为 2 档（D8）： exchange: inspection.status (direct) routing_key 格式: slot.{slotId}.{priority} priority ∈ {high, normal}

queue: inspection.status.s{slotId}.{priority} 示例: inspection.status.s073.high inspection.status.s073.normal 总队列数：

5.2 队列类型选择

Quorum Queue： Arguments: x-queue-type: quorum x-single-active-consumer: true x-max-length-bytes: 50000000 # 单队列限 50MB，背压保护 x-overflow: reject-publish-dlx x-dead-letter-exchange: inspection.dlx

队列声明（Java）: Map<String, Object> args = new HashMap<>(); args.put("x-queue-type", "quorum"); args.put("x-single-active-consumer", true); args.put("x-max-length-bytes", 50_000_000); args.put("x-overflow", "reject-publish-dlx"); args.put("x-dead-letter-exchange", "inspection.dlx"); channel.queueDeclare(queueName, true, false, false, args); 为什么 Quorum 而不是 Classic：

• Classic mirrored 已废弃
• Quorum 内置 HA，单节点崩溃不丢消息
• at-least-once 语义确定，不需要额外配置

⠀512 队列 × 3 节点 Raft 副本：每节点保存 ~1.5GB Raft log（按队列 50MB cap），完全可承受。

5.3 Channel 模型

第一版：每个活跃 slot consumer 一个独立 channel。 Worker 持有 50 个 slot × 2 优先级 = 100 channel 单 connection 默认上限 2047，完全够用 ack 路径清晰：delivery_tag 在自己的 channel 上 ack，无 frame interleaving 风险 Connection 配置： channelMax: 2048 heartbeat: 30 秒 automaticRecovery: true networkRecoveryInterval: 5 秒 topologyRecovery: true # 自动恢复 queue/binding

5.4 prefetch 策略

high queue: prefetch = 100 normal queue: prefetch = 300 原因：

• prefetch 过大 → Worker 本地堆积，故障恢复慢，OOM 风险
• prefetch 过小 → 网络往返开销占主导，吞吐下降
• 高优先级消息需要快速消费完成 → 较小 prefetch
• 普通消息允许更大批量 → 大 prefetch

⠀总 in-flight 限制：单 Worker (100 + 300) × 50 slot = 20000 条 in-flight。按单条 500B = 10MB 内存，可接受。

5.5 mailbox 与 SlotProcessor

mailbox 容量： high mailbox: 上限 = prefetch_high = 100 normal mailbox: 上限 = prefetch_normal = 300 满了之后：consumer 不再 ack，prefetch 自动卡住，背压传到 RabbitMQ。 SlotProcessor 调度： Worker 工作线程数 = max(物理核 × 1.5, 16)

调度策略：

1. 优先调度 mailbox 非空且未在 running 的 slot
2. 同 slot 永远只有一个线程在 drain
3. 老化保护：超过 30 秒未被调度的 slot 优先级提升
4. drain 时按 high → normal 顺序取消息

每个 drain 周期： 从 high 取最多 50 条 若 high 不足 50，从 normal 补到 200 条 执行 batch commit

6. 数据库设计

6.1 升级路径（D9 — P0）

当前 PG11 → PG16：

6.2 表结构总览

6.3 device_status_current 写入优化

-- fillfactor 留 30% 空间给 HOT update，避免索引膨胀 CREATE TABLE device_status_current (...) WITH (fillfactor=70);

-- autovacuum 加速 ALTER TABLE device_status_current SET ( autovacuum_vacuum_scale_factor = 0.05, autovacuum_vacuum_cost_limit = 2000 ); 批量写入策略： Worker 累积一批 (max 200 条 或 max 200ms) 后：

1. 应用侧按 device_id 去重，保留 seq 最大那条
2. 用 UNNEST 一次性发出 INSERT INTO device_status_current SELECT * FROM UNNEST(...) ON CONFLICT ... WHERE ...
3. 实际写入耗时：~10-30ms
4. 期间不持有任何长锁

6.4 history 治理

分区： CREATE TABLE device_status_history ( ts TIMESTAMPTZ NOT NULL, device_id VARCHAR(64) NOT NULL, seq BIGINT NOT NULL, payload JSONB NOT NULL, PRIMARY KEY (ts, device_id, seq) ) PARTITION BY RANGE (ts);

-- 每天分区，保留策略由 device priority 决定 保留策略：

6.5 rollup 表（前端报表入口）

CREATE TABLE device_status_rollup_1m ( minute_bucket TIMESTAMPTZ NOT NULL, device_id VARCHAR(64) NOT NULL, sample_count INT NOT NULL, state_changes INT NOT NULL, last_state VARCHAR(32), last_payload JSONB, PRIMARY KEY (minute_bucket, device_id) ) PARTITION BY RANGE (minute_bucket); 由 Worker 在每个 batch commit 时增量更新（同一事务内）。

7. 优先级机制（端到端）

7.1 设备分级

ALTER TABLE device_resource ADD COLUMN priority SMALLINT NOT NULL DEFAULT 2; -- 1: CRITICAL 2: NORMAL 3: LOW CRITICAL/HIGH 简化合并为 priority=1，对应 MQ high 队列。

7.2 IS 源头降频（最重要的优先级控制）

function inspectionLoop(): for device in deviceList: skipReason = checkBackpressure(device) if skipReason == "BACKPRESSURE_CRITICAL": if device.priority != 1: continue # 仅核心设备 elif skipReason == "BACKPRESSURE_HIGH": if device.priority == 3 and currentSlot % 4 != 0: continue # 低优先级设备 1/4 频率 # ... else 正常巡检

    msg = inspect(device)
    outbox.append(msg)

7.3 MQ 队列拆分

slot.{slotId}.high 与 slot.{slotId}.normal，按 device.priority 路由。

7.4 Worker 加权消费 + 防饥饿

function pickNextBatch(slotProcessor): high = slotProcessor.highMailbox.size() normal = slotProcessor.normalMailbox.size()

# 防饥饿：normal mailbox 最旧消息超过 60 秒，强制处理 normal
if normal > 0 and normalOldestAge() > 60s:
    return drain(normalMailbox, 200)

# 默认权重 high : normal = 3 : 1
if high > 0:
    return drain(highMailbox, min(150, high)) + drain(normalMailbox, 50)
else:
    return drain(normalMailbox, 200)

7.5 DB 写入降级

DB 写入异常率 > 30%（3 分钟窗口）→ 进入降级模式：

• device_status_current: 必写
• alert_event: 必写
• alert_current: 必写
• device_status_history: 仅核心设备写，其它丢弃（IS 端已写 outbox 不会真丢）
• rollup: 暂停，恢复后由后台补算（rollup 是从 history 派生的）

8. 容量规划（按 SLO 反推）

8.1 必须先明确的 SLO

8.2 三种规模档位

档位 A（POC / 小规模）：

8.3 IS 探针节点估算

9. 落地路线图

第 0 阶段：基础设施升级（与第一阶段并行，P0）

1. PostgreSQL 11 → 16 升级路径验证 2. RabbitMQ Quorum Queue 选型与部署 3. 监控系统（Prometheus + Grafana + OpenTelemetry）部署

第 1 阶段：架构纠偏（4 周）

1. 固定 TOTAL_SLOTS=256，建立 device → slot 稳定映射 2. 实现五版本号体系（resource / plan / slot_map / lease_epoch / schema） 3. IS 端去除对 lease_epoch 的感知 4. message_id 幂等键 + message_dedup 表 5. device_status_current 加 writer_epoch 列与 fencing UPSERT 模板（SDK 化） 6. Worker 改为按 slot_lease 消费 7. SlotProcessor 内部 mailbox 实现 8. IS 端 SQLite WAL outbox

第 2 阶段：调度闭环（4 周）

1. slot_lease 表 + worker_heartbeat 表 2. Web Leader 选主（PG advisory lock） 3. Worker 状态机（REGISTERED → WARMING → ACTIVE → DRAINING → DEAD → QUARANTINE） 4. Slot 状态机（FREE → ACTIVE → MOVING → DRAINING_OLD → LOADING_NEW） 5. rebalance 算法（权重计算、阈值触发、冷却） 6. 完整生命周期场景闭环测试（A/B/C/D/E/F/G 七场景） 7. derived_snapshot 持久化与加载

第 3 阶段：优先级与降级（3 周）

1. 设备 priority 字段 2. IS 端按优先级降频策略 3. slot.high / slot.normal 队列 4. Worker 加权消费 + 防饥饿 5. DB 写入降级模式 6. RabbitMQ / DB 故障演练

第 4 阶段：数据治理与告警重构（3 周）

1. alert_event / alert_current 重构 2. history 时间分区 3. rollup_1m 聚合 4. history 保留策略（按 priority） 5. 慢 SQL 治理与索引优化

第 5 阶段：压测与上线（2 周）

1. 1x / 2x / 5x 流量压测 2. 全部 §10 验收场景跑通 3. 灰度发布

10. 验收标准（必须全部通过才能上线）

11. 设计不变量（每次评审用这套校验）

INV-1: IS 投递的消息中携带 schema_version、resource_version、plan_version、slot_map_version INV-2: IS 永远不感知 slot_lease_epoch INV-3: Worker 扩缩容只改 slot_lease_epoch，不改其它任何版本 INV-4: device → slot 映射变更（slot_map_version 递增）走专门的"重分片协议" INV-5: 同一 device 的 (cycle_id, seq) 在 IS 端单调递增，永不回退 INV-6: message_id 全局唯一，IS 重启/重放后保持同值 INV-7: schema_version 升级时新旧 Worker 必须并行支持至少两个版本 INV-8: SlotProcessor 内不允许有任何"未持久化的状态" INV-9: 每次 batch commit 必须把派生状态作为快照写入 derived_snapshot INV-10: Worker 接管 slot 时加载所有受影响 device 的 derived_snapshot INV-11: IS 不允许"先发 MQ 后写 outbox"，必须先持久化 INV-12: outbox 的 message_id 必须能在 IS 重启后被重新构造 INV-13: outbox retention ≥ MQ 消息 TTL + 安全余量 INV-14: DB 中的 slot_lease 是事实源，Redis 仅为加速通知 INV-15: Worker 每次 batch commit 前必须确认本地 lease 缓存年龄 < 5 秒 INV-16: Worker 收到不属于自己的 slot 的 delivery 时立即 cancel + nack INV-17: 写入 device_status_current 必须使用标准 fencing UPSERT 模板 INV-18: alert_current 是 alert_event 的物化视图，不允许独立更新 INV-19: ack 必须在 DB commit 成功之后 INV-20: history 表写入是 append-only，依赖 PARTITION 治理而非 DELETE

12. 与原设计的对比（一句话总结）