cy-plugin-mqtt

基于 mqtt 封装的 MQTT 客户端插件，可同时用于：

• 浏览器端（WebSocket：ws / wss）
• Koa / Express 等 Node 服务端（可用 mqtt / mqtts，也可继续用 ws / wss）

v2.0.0 破坏性变更：peer 依赖 mqtt 已从 v4 升级为 mqtt.js v5（^5.2.0，建议 ^5.15.0）。业务项目需同步升级 mqtt。

✨ 特性

• 封装 mqtt，提供更友好的 API
• 内置 typed 事件发布/订阅（on / off / once / offAll）
• 自定义指数退避重连（默认关闭 mqtt.js 内置重连）
• 支持 ws / wss / mqtt / mqtts（浏览器仅 ws / wss）
• 通过 debug 控制调试日志（error 始终输出）
• ESM / CJS 双格式输出，支持 TypeScript 类型提示

📦 依赖与安装

mqtt 为 peerDependencies，不会打进本包，需在业务项目中自行安装（建议 mqtt@^5.15.0）。

浏览器端安装

npm install cy-plugin-mqtt mqtt@^5.15.0

Koa / Express 服务端安装

npm install cy-plugin-mqtt mqtt@^5.15.0

支持的 URL 示例

url 支持两种写法：

1. 完整 URL（已含 ://）→ 原样使用，此时 protocol / wss 不再二次拼接
2. host/path → 按 protocol（或旧字段 wss）自动拼成 protocol://url

protocol 可选值：ws | wss | mqtt | mqtts
未带 :// 且未指定协议时，默认使用 ws。

环境与协议对照

| 环境 | 可用协议 | 说明 | | --------------------- | ------------------------------- | ---------------------------------- | | 浏览器 | 仅 ws / wss | 使用 mqtt:// / mqtts:// 会抛错 | | Koa / Express（Node） | mqtt / mqtts / ws / wss | 推荐 TCP：mqtt:// / mqtts:// |

1. 完整 URL（推荐）

| 协议 | 示例 | 环境 | | ------- | ----------------------------------- | ------------- | | ws | ws://192.168.1.1:8083/mqtt | 浏览器 / Node | | ws | ws://broker.example.com:8083/mqtt | 浏览器 / Node | | wss | wss://192.168.1.1:8084/mqtt | 浏览器 / Node | | wss | wss://broker.example.com/mqtt | 浏览器 / Node | | mqtt | mqtt://127.0.0.1:1883 | 仅 Node / Koa | | mqtt | mqtt://broker.example.com:1883 | 仅 Node / Koa | | mqtts | mqtts://127.0.0.1:8883 | 仅 Node / Koa | | mqtts | mqtts://broker.example.com:8883 | 仅 Node / Koa |

new CyMqtt({ url: 'ws://192.168.1.1:8083/mqtt', username, password });
new CyMqtt({ url: 'wss://broker.example.com:8084/mqtt', username, password });
new CyMqtt({ url: 'mqtt://127.0.0.1:1883', username, password }); // Node
new CyMqtt({ url: 'mqtts://broker.example.com:8883', username, password }); // Node

2. host/path + protocol

| url | protocol | 最终结果 | 环境 | | ------------------------- | ----------------- | -------------------------------- | ------- | | 192.168.1.1:8083/mqtt | 不传（默认 ws） | ws://192.168.1.1:8083/mqtt | 两端 | | 192.168.1.1:8083/mqtt | 'ws' | ws://192.168.1.1:8083/mqtt | 两端 | | 192.168.1.1:8084/mqtt | 'wss' | wss://192.168.1.1:8084/mqtt | 两端 | | 127.0.0.1:1883 | 'mqtt' | mqtt://127.0.0.1:1883 | 仅 Node | | 127.0.0.1:8883 | 'mqtts' | mqtts://127.0.0.1:8883 | 仅 Node | | broker.example.com:1883 | 'mqtt' | mqtt://broker.example.com:1883 | 仅 Node |

new CyMqtt({ url: '192.168.1.1:8083/mqtt', protocol: 'ws', username, password });
new CyMqtt({ url: '192.168.1.1:8084/mqtt', protocol: 'wss', username, password });
new CyMqtt({ url: '127.0.0.1:1883', protocol: 'mqtt', username, password });
new CyMqtt({ url: '127.0.0.1:8883', protocol: 'mqtts', username, password });

3. 兼容旧字段 wss

| url | wss | 最终结果 | | ---------------- | -------------- | ---------------------- | | host:8083/mqtt | 不传 / false | ws://host:8083/mqtt | | host:8084/mqtt | true | wss://host:8084/mqtt |

new CyMqtt({ url: 'host:8084/mqtt', wss: true, username, password });
// 等价于 protocol: 'wss'

若同时传 protocol 与 wss，以 protocol 优先。

4. 浏览器禁止的写法（会抛错）

// ❌ 浏览器环境不可用
new CyMqtt({ url: 'mqtt://127.0.0.1:1883', username, password });
new CyMqtt({ url: 'mqtts://127.0.0.1:8883', username, password });
new CyMqtt({ url: '127.0.0.1:1883', protocol: 'mqtt', username, password });

浏览器端

注意事项

1. 协议只能是 ws / wss，不能使用 mqtt://、mqtts://。
2. 必须连接成功后再 subscribe / publish：监听 CONNECT，或先判断 mqtt.isConnected；未连接调用会触发 ERROR（「尚未连接，无法…」）。
3. 页面卸载或路由销毁时务必调用 destroy()，避免连接与定时器泄漏。
4. 建议配置稳定的 connectOption.clientId。
5. mqtt.js v5 起多数打包器可直接 import mqtt from 'mqtt'；若仍报 net / tls 等错误，再将 mqtt 别名到 mqtt/dist/mqtt.esm.js。

示例

import CyMqtt, { enumMqttEvent } from 'cy-plugin-mqtt';

const mqtt = new CyMqtt({
  url: '192.168.1.1:8083/mqtt',
  protocol: 'ws', // 或 'wss'
  username: 'admin',
  password: '123456',
  debug: false,
  subscribeTopic: 'device/',
  publishTopic: 'device/',
  qos: 0,
  connectOption: {
    clientId: 'web-client-001',
    clean: true,
    maxCount: Infinity,
    interval: 5_000,
    maxInterval: 60_000
  }
});

mqtt.connect();

// 方式一：在 CONNECT 回调里再订/发（推荐）
mqtt.on(enumMqttEvent.CONNECT, () => {
  if (!mqtt.isConnected) {
    return;
  }
  mqtt.subscribe('device123');
  mqtt.publish({ text: 'hello' });
});

mqtt.on(enumMqttEvent.MESSAGE, ({ topic, data }) => {
  console.log('收到消息:', topic, data);
});

mqtt.on(enumMqttEvent.ERROR, (err) => {
  console.error('MQTT 错误:', err);
});

// 方式二：业务任意时机发布前先判断
function sendMessage(payload: unknown) {
  if (!mqtt.isConnected) {
    console.warn('MQTT 未连接，暂不可发布');
    return;
  }
  mqtt.publish(payload);
}

// 页面卸载时释放资源
window.addEventListener('pagehide', () => {
  mqtt.destroy();
});

Koa / Express 服务端

注意事项

1. 推荐使用 mqtt:// / mqtts://（TCP），性能通常优于 WebSocket；若 Broker 只开了 WS，再用 ws / wss。
2. 使用进程级单例，不要在每个请求里 new CyMqtt()，否则会连接风暴、clientId 冲突。
3. 建议固定 connectOption.clientId（如 koa-server-1）。
4. 在进程退出 / 服务关闭时调用 destroy()（SIGTERM、SIGINT、app.close 等）。
5. Express 与 Koa 用法相同：共享同一个 MQTT 实例即可。
6. 请求里发布前必须判断 mqtt.isConnected；订阅建议放在 CONNECT 回调中。

Koa 示例

import Koa from 'koa';
import CyMqtt, { enumMqttEvent } from 'cy-plugin-mqtt';

const app = new Koa();

// 进程级单例
const mqtt = new CyMqtt({
  url: 'mqtt://127.0.0.1:1883',
  username: 'admin',
  password: '123456',
  debug: false,
  subscribeTopic: 'device/',
  publishTopic: 'device/',
  connectOption: {
    clientId: 'koa-server-1',
    clean: true,
    maxCount: Infinity,
    interval: 5_000,
    maxInterval: 60_000
  }
});

mqtt.connect();

mqtt.on(enumMqttEvent.CONNECT, () => {
  if (!mqtt.isConnected) {
    return;
  }
  mqtt.subscribe('device123');
});

mqtt.on(enumMqttEvent.MESSAGE, ({ topic, data }) => {
  console.log('[mqtt]', topic, data);
});

mqtt.on(enumMqttEvent.ERROR, (err) => {
  console.error('[mqtt] error', err);
});

app.use(async (ctx) => {
  if (ctx.path === '/publish') {
    if (!mqtt.isConnected) {
      ctx.status = 503;
      ctx.body = { ok: false, message: 'mqtt disconnected' };
      return;
    }
    mqtt.publish({ from: 'koa', at: Date.now() });
    ctx.body = { ok: true };
    return;
  }
  ctx.body = { connected: mqtt.isConnected };
});

const server = app.listen(3000);

function shutdown() {
  mqtt.destroy();
  server.close(() => process.exit(0));
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

Express 示例

import express from 'express';
import CyMqtt, { enumMqttEvent } from 'cy-plugin-mqtt';

const app = express();

const mqtt = new CyMqtt({
  url: 'mqtt://127.0.0.1:1883',
  username: 'admin',
  password: '123456',
  connectOption: {
    clientId: 'express-server-1'
  }
});

mqtt.connect();

mqtt.on(enumMqttEvent.CONNECT, () => {
  if (!mqtt.isConnected) {
    return;
  }
  mqtt.subscribe('device123');
});

mqtt.on(enumMqttEvent.MESSAGE, ({ topic, data }) => {
  console.log('[mqtt]', topic, data);
});

mqtt.on(enumMqttEvent.ERROR, (err) => {
  console.error('[mqtt] error', err);
});

app.post('/publish', (_req, res) => {
  if (!mqtt.isConnected) {
    res.status(503).json({ ok: false, message: 'mqtt disconnected' });
    return;
  }
  mqtt.publish({ from: 'express', at: Date.now() });
  res.json({ ok: true });
});

const server = app.listen(3000);

function shutdown() {
  mqtt.destroy();
  server.close(() => process.exit(0));
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

API

| 方法 | 说明 | | -------------------------------- | ---------------------------------------------------------------------- | | connect() | 连接 Broker | | disconnect() | 断开（可再次 connect；订阅会保留并在重连后恢复） | | destroy() | 销毁实例（清订阅、清监听、不可再连） | | subscribe(id, prefix?) | 订阅 | | unsubscribe(id, prefix?) | 取消订阅 | | publish(msg, topic?, options?) | 发布；对象默认 JSON；string / 二进制原样；{ raw: true } 跳过序列化 | | on / off / once / offAll | 事件监听 | | isConnected | 是否已连接 |

常用事件：enumMqttEvent.CONNECT / RECONNECT / CLOSE / ERROR / MESSAGE / SUBSCRIBE / UNSUBSCRIBE。

MESSAGE 回调参数为 { topic, data }。