基于JavaScript的MQTT实时通信应用开发实战记录

const mqttClient = new MqttClient('mqtt://broker.emqx.io');
// 连接并订阅
mqttClient.connect().then(() => {
    mqttClient.subscribe('home/sensor', (topic, message) => {
        console.log(`[${topic}] ${message}`);
    });
    // 发布消息
    mqttClient.publish('home/light', 'ON');
});
// 页面卸载时断开连接
window.addEventListener('beforeunload', () => mqttClient.disconnect());

/**
 * 
 * 调用方式如下
 * const mqttClient = new MqttClient('mqtt://your-broker-url');
 * mqttClient.connect();
 * mqttClient.subscribe('your/topic', (topic, message) => {
 * console.log(`${topic}: ${message}`);
 * });
 * 
 * 在页面销毁时断开连接，释放资源（防止骂娘）
 * mqttClient.disconnect();
 * */
   import mqtt from 'mqtt';
// 定义MqttClient类
class MqttClient {
	// 构造函数，接收broker的URL和客户端ID（可选）
	constructor(brokerUrl, clientId = `client-${Math.random().toString(16).substr(2, 8)}`, username = '', password = '') {
		this.brokerUrl = brokerUrl; // 存储broker的URL
		this.clientId = clientId; // 存储客户端ID，如果没有提供则生成一个随机的
		this.client = null; // 初始化mqtt客户端为null
		this.reconnectCount = 0; // 添加重连计数器
		this.maxReconnectAttempts = 5; // 设置最大重连次数
		this.username = username;
		this.password = password;
	}
	/**
	 * 连接到MQTT broker的方法，返回Promise
	 */
	connect() {
		return new Promise((resolve, reject) => {
			const options = {
				clientId: this.clientId,
				clean: true,
				connectTimeout: 4000,
				reconnectPeriod: 1000,
				username: this.username, // 添加用户名配置
				password: this.password // 添加密码配置
			};
			this.client = mqtt.connect(this.brokerUrl, options);
			this.client.on('connect', () => {
				console.log('M_connected');
				this.reconnectCount = 0; // 连接成功后重置计数器
				resolve(true);
			});
			this.client.on('error', (error) => {
				console.error('M_error:', error);
				reject(error);
			});
			this.client.on('close', () => {
				console.log('M_connection closed');
				reject(new Error('Connection closed'));
			});
			this.client.on('offline', () => {
				console.log('M_client offline');
				reject(new Error('Client offline'));
			});
			this.client.on('reconnect', () => {
				this.reconnectCount++;
				console.log(`M_reconnecting... (Attempt ${this.reconnectCount}/${this.maxReconnectAttempts})`);
				if (this.reconnectCount >= this.maxReconnectAttempts) {
					this.client.end(); // 达到最大重连次数后断开连接
					reject(new Error('Max reconnection attempts reached'));
				}
			});
		});
	}
	/**
	 * 订阅主题的方法，接收主题和回调函数作为参数
	 * @param {string} topic 
	 * @param {fun} callback 
	 */
	subscribe(topic, callback) {
		this.client.subscribe(topic, { qos: 1 }, (error) => { // 使用qos 1保证消息至少被传递一次，2 网络开销较大，不推荐用 0
			if (error) {
				console.error('Subscribe error:', error);
			} else {
				// 监听消息事件，当收到消息时调用回调函数，receivedTopic与订阅主题匹配的主题
				this.client.on('message', (receivedTopic, message) => {
					callback(receivedTopic, message.toString()); // 调用回调函数处理消息
				});
			}
		});
	}
	/**
	 * 发布消息到指定主题的方法，接收主题、消息和qos（可选）作为参数
	 * @param {string} topic 
	 * @param {string|Buffer} message 
	 * @param {number} qos 
	 */
	publish(topic, message, qos = 1) {
		this.client.publish(topic, message, { qos }, (error) => { // 使用指定的qos发布消息
			if (error) {
				console.error('Publish error:', error);
			} else {
				console.log(`Published message to topic: ${topic}`);
			}
		});
	}
	/**
	 * 断开与MQTT broker连接的方法
	 */
	disconnect() {
		// this.client.end();
		if (this.client?.connected) {
			this.client.end(); // 断开连接
		}
	}
}
export default MqttClient;