微信小程序使用MQTT问题记录

这里环境是 uniapp vue3 typescript

很多国内轻量 IOT 应用都选择开发小程序做入口,那么肯定会集成 MQTT 框架，通常是选用 mqtt.js ,但是这个库是给 web 设计的, 很多环境微信小程序中没有,因此会遇到一些问题,常见的有

1. 直接使用无法连接上MQTT

__2. 如果手动导入引用 mqtt.min.js __

初发现可以使用,也是网上常见的解决办法,什么用旧版本 4.0.2 之类的解决办法 . 但是深度测试会发现,很多 MQTT__生命周期不会回调！！！__ 这种属于深坑了,会导致一些应用 BUG 影响用户体验.

所以我们的解决思路还是以解决官方包使用无效的方向为主

解决方案

1. 导入额外依赖

首先看下官方文档在关于小程序支持上有说明

要增加这些部分

1
2
3
4
5
6
import 'abortcontroller-polyfill/dist/abortcontroller-polyfill-only' // import before mqtt.
import 'esbuild-plugin-polyfill-node/polyfills/navigator'
const mqtt = require("mqtt");
const client = mqtt.connect("wxs://test.mosquitto.org", {
  timerVariant: 'native' // more info ref issue: #1797
});

实际操作中发现,前面的 import 如果只是在同一个文件中导入顺序有限于mqtt.js ,那么实际上是无效的.

所以只能放到 main.ts 的最上方

第一行导入后也需要执行 npm install 能解决 AbsortController 缺失的报错 ,然而第二行还是会报错,我们直接去除. navigator这边查看源码后,只是一个简单的对象, 我们手动加入到源码即可

var navigator={deviceMemory:8,hardwareConcurrency:8,language:"en-Us"}; 这段代码手动添加到 node_modules/mqtt/dist/mqtt.esm.js最前面 至此 navigator 问题解决

2. 配置问题

现在会遇到 WebSocket is not a constructor 的错误,或者不报错.这里是因为小程序环境不传入 websocket对象的话默认new的是有问题的,所以我们要手动传入一个, 这里重点是 timerVariant,forceNativeWebSocket,createWebsocket 这三个参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
private buildClientOptions() : IClientOptions {
		return {
			clientId: this.clientId,
			username: this.userId,
			password: this.boundAccessToken,
			clean: false,
			keepalive: KEEP_ALIVE_SECONDS,
			reconnectPeriod: RECONNECT_PERIOD_MS,
			connectTimeout: CONNECT_TIMEOUT_MS,
			protocolVersion: 5,
			resubscribe: true,
			properties: {
				sessionExpiryInterval: SESSION_EXPIRY_SECONDS,
			},
			will: {
				topic: this.willTopic,
				payload: JSON.stringify({ foreground: false }),
				qos: 1,
				retain: false,
			},
			timerVariant: 'native',
			forceNativeWebSocket: true,
			createWebsocket: (url : string, websocketSubProtocols : string[], options : IClientOptions) : WebSocket => {
				const socketTask = wx.connectSocket({
					//url: url.replace(/^ws:/, 'wss:'), // 微信小程序必须使用 wss
					url, // 微信小程序必须使用 wss
					protocols: websocketSubProtocols,
					header: {
						'content-type': 'application/json',
					},
				});

				const listeners : Record<string, Function[]> = {
					open: [],
					close: [],
					error: [],
					message: [],
				};

				let readyState = 0; // CONNECTING = 0
				let binaryType = 'arraybuffer';

				// 消息队列缓存
				const messageQueue : (string | ArrayBuffer)[] = [];

				socketTask.onOpen(() => {
					readyState = 1; // OPEN
					listeners.open.forEach(fn => fn());

					// 发送缓存的消息（确保已经是 string | ArrayBuffer）
					while (messageQueue.length > 0) {
						const msg = messageQueue.shift();
						if (msg === undefined) continue;

						socketTask.send({
							data: msg, // 直接发送，无需转换
							fail: err => console.error('Queued message send error:', err),
						});
					}
				});

				socketTask.onClose(res => {
					readyState = 3; // CLOSED
					listeners.close.forEach(fn => fn(res));
				});

				socketTask.onError(res => {
					readyState = 3;
					listeners.error.forEach(fn => fn(res));
				});

				socketTask.onMessage(res => {
					listeners.message.forEach(fn => fn({ data: res.data }));
				});

				const socketLike : WebSocket = {
					get readyState() {
						return readyState;
					},

					get binaryType() {
						return binaryType;
					},
					set binaryType(type) {
						binaryType = type;
					},

					send(data : string | ArrayBufferLike | Blob | ArrayBufferView) {
						let payload : string | ArrayBuffer;

						try {
							payload = toStandardArrayBuffer(data);
						} catch (err) {
							console.error('Data conversion error:', err);
							return;
						}

						if (readyState === 1) {
							socketTask.send({
								data: payload,
								fail: (err) => console.error('Send error:', err),
							});
						} else {
							console.warn('Socket not open yet, message queued');
							messageQueue.push(payload);
						}
					},

					close(code ?: number, reason ?: string) {
						socketTask.close({ code, reason });
					},

					addEventListener(type : string, listener : Function) {
						if (listeners[type]) {
							listeners[type].push(listener);
						}
					},

					removeEventListener(type : string, listener : Function) {
						if (listeners[type]) {
							listeners[type] = listeners[type].filter(fn => fn !== listener);
						}
					},

					// 兼容 mqtt.js 的事件写法（socket.onopen = ...）
					set onopen(fn : Function) {
						listeners.open = [fn];
					},
					set onclose(fn : Function) {
						listeners.close = [fn];
					},
					set onerror(fn : Function) {
						listeners.error = [fn];
					},
					set onmessage(fn : (arg0 : any) => any) {
						listeners.message = [(e : any) => fn(e)];
					},

					// 类型断言，表示这是符合 WebSocket 类型
				} as any;

				return socketLike;
			}
		}
	}
1
const client = mqtt.connect(MQTT_URL, this.buildClientOptions())

解决!