HarmonyOS APP开发—"蜂巢"工业设备运维监控App，需要用到这个库

1000 台设备 × 每秒 1 次上报 = 每秒 1000 条消息。HTTP 轮询在这个量级直接崩盘。@ohos/mqtt 的长连接 + 发布订阅一个客户端就能扛住，QoS 分级还能让"指标丢两条无所谓、告警一条都不能丢"变成配置项而不是奢望。

📦 仓库地址 ｜ ohpm install @ohos/mqtt

---

场景

"蜂巢"接入车间上千台设备：

• 上行：每台设备每秒上报温度、振动、电流
• 下行：运维批量下发控制指令（启停、参数调整）
• 告警：设备异常立即推送，一条都不能丢
• 状态：设备掉线要马上知道

关键难点是量级和可靠性分级——指标数据可以丢，告警绝不能丢，这两种需求要在一套连接里同时满足。

---

为什么是 MQTT

需求	HTTP 轮询	WebSocket	MQTT
千级设备	❌ 每秒千次请求	⚠️ 需自建路由	✅ 主题订阅
可靠性分级	❌	❌	✅ QoS 0/1/2
掉线检测	❌	❌ 手写心跳	✅ 遗嘱机制
一对多下发	❌	⚠️ 手写分发	✅ 通配符主题
离线消息	❌	❌	✅ retained + 会话保持
报文开销	大	中	✅ 极小

---

关键实现

主题设计要有层级

层级设计决定了通配符订阅的能力：

factory/{厂区}/line{产线}/device/{设备ID}/metrics   ← 指标
factory/{厂区}/line{产线}/device/{设备ID}/alarm     ← 告警
factory/{厂区}/line{产线}/device/{设备ID}/cmd       ← 指令
factory/{厂区}/line{产线}/device/{设备ID}/status    ← 状态

import { MqttClient, MqttConnectOptions } from '@ohos/mqtt'

client.on('connect', () => {
  client.subscribe('factory/#/metrics', { qos: 0 })   // 指标：QoS 0，实时性优先
  client.subscribe('factory/#/alarm', { qos: 1 })     // 告警：QoS 1，绝不能丢
  client.subscribe('factory/#/status', { qos: 1 })    // 状态：QoS 1
})

// 高频消息，回调里只做轻量处理
client.on('message', (topic: string, payload: ArrayBuffer) => {
  const parts = topic.split('/')
  const deviceId = parts[3]
  const type = parts[parts.length - 1]
  const data = JSON.parse(new util.TextDecoder().decodeToString(new Uint8Array(payload)))

  if (type === 'metrics') {
    this.metrics.set(deviceId, data)
    if (data.temperature > 85 || data.vibration > 12) this.handleAlarm(deviceId, data)
  } else if (type === 'alarm') {
    this.handleAlarm(deviceId, data)
  }
})

QoS 分级策略：

消息类型	QoS	理由
指标数据	0	量大，丢一两条无所谓，实时性优先
告警	1	绝不能丢
设备状态	1	掉线/上线必须准确
控制指令	2	重复执行可能危险，要恰好一次

遗嘱 + 延迟，避免告警风暴

const options: MqttConnectOptions = {
  cleanSession: false,
  keepAliveInterval: 30,
  will: {
    topic: `factory/line1/device/${DEVICE_ID}/status`,
    payload: JSON.stringify({ online: false }),
    qos: 1,
    properties: {
      willDelayInterval: 30,   // ✅ 延迟 30 秒再发遗嘱
    },
  },
}

willDelayInterval 在车间特别有用：网络偶尔抖动，设备可能几秒后就重连了。设 30 秒延迟能避免"抖动 → 报掉线 → 马上又上线"的告警风暴。

---

两个容易踩的点

1. 高频发布要分批节流

瞬间发上千条消息会给 Broker 和客户端 native 层造成压力。运维点一下"全线停机"时：

async function broadcastCommand(devices: string[], cmd: string) {
  const BATCH = 50, INTERVAL = 100   // ms

  for (let i = 0; i < devices.length; i += BATCH) {
    devices.slice(i, i + BATCH).forEach((id) => {
      const payload = JSON.stringify({ cmd, timestamp: Date.now() })
      if (payload.length > 256 * 1024) return   // 单条大小校验
      client.publish(`factory/line1/device/${id}/cmd`, payload, { qos: 1 })
    })
    await new Promise((r) => setTimeout(r, INTERVAL))   // 批间喘息
  }
}

分批还有个附带好处：能顺带做进度反馈，运维知道指令下发到哪了。

2. keepAliveInterval 要匹配业务节奏

设备每 3 秒上报一次，如果 keepAliveInterval 设成 3 或更小，Broker 会因为"没在心跳周期内收到包"而断开连接。

const REPORT_INTERVAL = 3   // 秒
const options = {
  keepAliveInterval: REPORT_INTERVAL * 3,   // ✅ 设为业务间隔的 2~3 倍
  cleanSession: false,
}

---

用 V5 属性解决告警堆积

运维第二天上班，不该被昨天的告警淹没：

client.publish('factory/line1/device/D042/alarm', JSON.stringify({ type: 'overheat', temp: 92.5 }), {
  qos: 1,
  properties: {
    messageExpiryInterval: 3600,      // 1 小时后过期，不再推送
    contentType: 'application/json',
    userProperties: {                 // 打标签，订阅端可按此过滤
      severity: 'critical',
      line: 'line1',
      shift: 'A',
    },
  },
})

userProperties 的好处是不用改 payload 结构就能加业务元信息，订阅端过滤时也不用解析整个 JSON。

---

小结

"蜂巢"这个场景里，@ohos/mqtt 解决了三件事：通配符订阅让一个客户端扛住上千台设备、QoS 分级让指标和告警各得其所、遗嘱机制零代码实现掉线检测。

工业 IoT 是 MQTT 的极限压力场。高频发布记得分批节流，心跳间隔记得留足余量——这两条做到了，稳定性基本就有保障了。