MQTT Service Library

一个功能完整的 MQTT 服务库，提供监控、健康检查、自动重连等企业级功能。

特性

• 🔄 自动重连 - 连接断开时自动重连
• 📊 实时监控 - 连接状态和消息监控
• 🏥 健康检查 - 内置心跳检测机制
• 📦 消息缓冲 - 支持消息批量处理
• 🎯 类型安全 - 完整的 TypeScript 支持
• 🔧 灵活配置 - 丰富的配置选项
• 📈 性能指标 - 消息延迟监控

安装

npm install mqtt-service-lib
# 或
yarn add mqtt-service-lib
# 或
pnpm add mqtt-service-lib

快速开始

基本使用

import { MQTTService } from 'mqtt-service-lib';

// 创建 MQTT 服务实例
const mqttService = MQTTService.getInstance('mqtt://localhost:1883', {
  username: 'your-username',
  password: 'your-password',
  enableRetry: true,
  enableMetrics: true
});

// 开始监控
mqttService.monitorByIntegration({
  topics: ['sensor/data', 'device/status'],
  interval: 1000, // 1秒间隔批量处理消息
  onMessage: (messages) => {
    console.log('收到消息:', messages);
  },
  onConnectionStatus: (status) => {
    console.log('连接状态:', status);
  },
  unstableThreshold: 5000, // 5秒无消息认为不稳定
  enableDebug: true
});

// 发布消息
mqttService.publish('test/topic', 'Hello MQTT!', 1, false);

// 停止监控
mqttService.unmonitor();

高级配置

import { MQTTService, type MQTTServiceOptions } from 'mqtt-service-lib';

const options: MQTTServiceOptions = {
  username: 'admin',
  password: 'password123',
  enableRetry: true,        // 启用消息重试
  enableMetrics: true,      // 启用性能指标
  enableHealthCheck: true,  // 启用健康检查
  reconnectPeriod: 3000,    // 重连间隔
  connectTimeout: 5000,     // 连接超时
  keepalive: 60,           // 保活时间
};

const mqttService = MQTTService.getInstance('mqtt://broker.example.com:1883', options);

API 参考

MQTTService

构造函数

MQTTService.getInstance(url: string, options?: MQTTServiceOptions): MQTTService

方法

monitorByIntegration(options)

开始集成监控。

monitorByIntegration(options: MQTTMonitorOptions): void

参数:

• options.topics - 要订阅的主题数组
• options.interval - 消息批量处理间隔（毫秒），0 表示实时处理
• options.onMessage - 消息回调函数
• options.onConnectionStatus - 连接状态回调函数
• options.unstableThreshold - 不稳定阈值（毫秒）
• options.enableDebug - 是否启用调试日志

unmonitor()

停止监控并断开连接。

unmonitor(): void

publish(topic, message, qos?, retain?)

发布消息。

publish(topic: string, message: string, qos?: 0 | 1 | 2, retain?: boolean): void

unsubscribe(topics)

取消订阅主题。

unsubscribe(topics: string[]): void

isConnected()

检查是否已连接。

isConnected(): boolean

getStatus()

获取连接状态。

getStatus(): ConnectionStatus

类型定义

MQTTMessage

interface MQTTMessage {
  topic: string;        // 主题
  payload: string;      // 消息内容
  timestamp: number;    // 时间戳
  latency?: number;     // 延迟（毫秒）
}

ConnectionStatus

type ConnectionStatus = 'connecting' | 'connected' | 'unstable' | 'disconnected';

MQTTMonitorOptions

interface MQTTMonitorOptions {
  topics: string[];                    // 订阅的主题
  interval?: number;                   // 消息处理间隔
  onMessage: (data: MQTTMessage[]) => void;  // 消息回调
  onConnectionStatus?: (status: ConnectionStatus) => void;  // 状态回调
  unstableThreshold?: number;          // 不稳定阈值
  enableDebug?: boolean;               // 调试模式
}

MQTTServiceOptions

interface MQTTServiceOptions extends IClientOptions {
  username?: string;           // 用户名
  password?: string;           // 密码
  enableRetry?: boolean;       // 启用重试
  enableMetrics?: boolean;     // 启用指标
  enableHealthCheck?: boolean; // 启用健康检查
}

连接状态说明

• connecting - 正在连接
• connected - 已连接且稳定
• unstable - 已连接但不稳定（超过阈值时间未收到消息）
• disconnected - 已断开连接

开发

安装依赖

pnpm install

构建

pnpm build

测试

pnpm test

代码检查

pnpm lint

许可证

MIT License

贡献

欢迎提交 Issue 和 Pull Request！