自建 WS 推送（本周前几天）：
  服务器 ──必须维护──▶ 每个客户端的订阅表、状态、补传逻辑（全要自己写）

MQTT：
  设备 ──publish──▶ Broker（订阅表/QoS/离线消息全由 Broker 负责）◀──subscribe──▶ 应用
  你的服务端代码量断崖式下降

┌────────┐  publish "factory/line1/furnace-1/temp"  ┌────────┐
│ 加热炉  │ ──────────────────────────────────────▶ │        │
└────────┘            85.3 ℃                        │ Broker │  (Mosquitto/EMQX/
                                                      │        │   云厂商 IoT 平台)
┌────────┐  subscribe "factory/+/furnace-+/temp"    │        │
│ 监控页  │ ──────────────────────────────────────▶ │        │
└────────┘◀────── Broker 把匹配的消息推给订阅者 ──────┘
└────────┘

推荐结构：{站点}/{产线}/{设备}/{通道}

factory/line1/furnace-1/temperature     ✅ 温度数据
factory/line1/furnace-1/status          ✅ 设备状态
factory/line1/furnace-1/alarm           ✅ 告警事件
factory/line2/press-3/pressure

设计纪律：
1. 层级从粗到细（站点→产线→设备→通道），匹配通配符的威力
2. 全小写 + 连字符（别混下划线/驼峰）
3. 不要把数据放 Topic 里（"factory/temp/85.3" ❌——Topic 是地址不是内容）

// 单层通配符 +：匹配恰好一层
"factory/+/furnace-1/temp"
// 匹配 factory/line1/furnace-1/temp、factory/line2/furnace-1/temp
// 不匹配 factory/line1/furnace-2/temp

// 多层通配符 #：匹配任意层级（必须在末尾）
"factory/line1/#"
// 匹配 factory/line1 下所有消息（任何设备任何通道）

// 组合：大屏只关心温度告警
"factory/+/+/alarm"

发送方                          接收方
  │ ── PUBLISH(msg, id=42) ──▶  │   收到，处理
  │                              │ ── PUBACK(42) ──▶  │   ACK 在路上丢了！
  │ ✗ 没收到 PUBACK(42)          │
  │ ── PUBLISH(msg, id=42) 重发 ▶ │   又收到 id=42
                                  │   → 重复投递（业务层要按 id 去重）

温度振动流（1kHz）        → QoS 0（重传一帧旧数据反而是污染）
设备启停/告警            → QoS 1（宁可重复告警，不能漏）
远程设定值下发/配方下发   → QoS 2（执行两次 = 事故）

痛点：监控页 10:05 才打开，设备 10:00 发的"当前温度"已经过去了——
      新页面要傻等下一次推送（1s~更久）。

retain 消息：Broker 为每个 Topic 存最后一条 retain 消息，
新订阅者连上的瞬间立刻收到它。

// 发布端：标记 retain
client.publish("factory/line1/furnace-1/temp", "85.3", { retain: true });

// 任何新订阅者 connect 后立刻收到这条"温度 85.3"，不用等下一次变化

痛点：设备断电了，谁去告诉 Broker 和所有监控页"我下线了"？
      设备自己肯定来不及说。

LWT：设备连接时预先登记一条"遗嘱"——
     Broker 检测到设备异常掉线（keepalive 超时），替它发布这条遗嘱。

// 设备端：连接时登记遗嘱
const client = mqtt.connect("wss://broker.emqx.io:8084/mqtt", {
  clientId: "furnace-1",
  will: {                                // ⭐ 遗嘱
    topic: "factory/line1/furnace-1/status",
    payload: JSON.stringify({ online: false, ts: Date.now() }),
    qos: 1,
    retain: true,
  },
});
// 设备正常在线时自己发布 { online: true }（retain）
// 设备崩溃/断电 → Broker 代发遗嘱 { online: false } → 所有监控页立刻感知

npm install mqtt

import mqtt from "mqtt";

/**
 * 浏览器接入 MQTT（MQTT over WebSocket——浏览器不能裸连 TCP 1883）
 */
const client = mqtt.connect("wss://broker.emqx.io:8084/mqtt", {
  clientId: `panel-${Math.random().toString(16).slice(2, 8)}`,
  keepalive: 30,               // 心跳间隔（秒）——Broker 侧判死依据
  clean: false,                // false 保留会话：断线期间 QoS1 消息由 Broker 暂存
  reconnectPeriod: 2000,       // mqtt.js 内建重连（体会：库帮你做了 Day 58 的事）
});

client.on("connect", () => {
  console.log("已连接 Broker");
  // 通配符订阅：全厂所有设备的温度和告警
  client.subscribe(
    ["factory/+/+/temperature", "factory/+/+/alarm"],
    { qos: 1 },
    (err) => { if (!err) console.log("订阅成功"); }
  );
});

client.on("message", (topic: string, payload: Buffer) => {
  // 按 Topic 分发（MqttDataSource 的雏形）
  if (topic.endsWith("/temperature")) {
    const value = Number(payload.toString());
    console.log(`温度 ${topic} = ${value}`);
  } else if (topic.endsWith("/alarm")) {
    const alarm = JSON.parse(payload.toString());
    console.log("告警", alarm);
  }
});

client.on("offline", () => console.log("连接断开，自动重连中…"));

1. MQTTX 连同一个 Broker（wss://broker.emqx.io:8084/mqtt）
2. 发布：Topic = factory/line1/furnace-1/temperature，内容 = 85.3，QoS 1，勾选 retain
3. 回浏览器：onmessage 触发 ✅
4. 刷新浏览器页面：retain 消息立刻送达 ✅

import type { DataSource, PanelSnapshot } from "./types";
import mqtt from "mqtt";

/**
 * MQTT 数据源：实现与 MockDataSource 相同的 DataSource 契约
 * 面板代码零改动即可切换 —— 第 8 周架构承诺的又一次兑现
 */
export class MqttDataSource implements DataSource {
  private client = mqtt.connect("wss://broker.emqx.io:8084/mqtt", {
    keepalive: 30,
    reconnectPeriod: 2000,
  });
  private listeners: ((s: PanelSnapshot) => void)[] = [];

  constructor() {
    this.client.on("connect", () => {
      this.client.subscribe("factory/+/+/+", { qos: 1 });
    });
    this.client.on("message", (_topic, payload) => {
      // 生产实现：按 topic 聚合各设备数据，组装 PanelSnapshot 后分发
      // （聚合逻辑与 BOSS 战的 WsDataSource 同构，留到 Day 63 统一实现）
      this.listeners.forEach((l) => l(JSON.parse(payload.toString())));
    });
  }

  subscribe(listener: (s: PanelSnapshot) => void): () => void {
    this.listeners.push(listener);
    return () => { this.listeners = this.listeners.filter((l) => l !== listener); };
  }
  start(): void { /* mqtt.js 自动连接，无需额外启动 */ }
  stop(): void { this.client.end(true); }
}