使用 NodeJS 搭建 MQTT 服务端

引言

随着物联网（IoT）的迅速发展，MQTT协议因其轻量级和高效的特性，逐渐成为设备间数据传输的主流协议。本文将介绍如何使用NodeJS搭建一个基本的MQTT服务端，帮助开发者快速构建物联网应用。

什么是MQTT？

MQTT（Message Queuing Telemetry Transport）是一种发布/订阅（Publish/Subscribe）消息传输协议，专为低带宽、不可靠网络环境下的设备通信而设计。它允许客户端订阅感兴趣的主题，当有新消息时，服务端会推送给相关的订阅者。

准备工作

NodeJS：请确保安装了 NodeJS（推荐版本 >= 14.x）。
MQTT Broker库：我们将使用Mosca或Aedes库，它们是基于 NodeJS 的轻量级 MQTT broker 实现。
本次实战依然基于 MacOS

步骤1：创建项目并安装依赖

初始化一个新的NodeJS项目：

mkdir test-mqtt-server
cd test-mqtt-server
npm init -y

安装 Aedes 库（或 M）以及MQTT：
```
npm install aedes mqtt --save
```

步骤2：编写 MQTT 服务端代码

在项目根目录下创建server.js文件，编写MQTT服务端的基础代码：

const aedes = require('aedes')();
const server = require('net').createServer(aedes.handle);
const PORT = 1883;server.listen(PORT, function () {console.log(`MQTT server is running on port ${PORT}`);
});aedes.on('client', (client) => {console.log(`Client Connected: ${client.id}`);
});aedes.on('clientDisconnect', (client) => {console.log(`Client Disconnected: ${client.id}`);
});aedes.on('publish', (packet, client) => {console.log(`Message Published: ${packet.payload.toString()}`);
});

步骤3：运行并测试服务端

启动MQTT服务端：
```
node server.js
```
使用MQTT客户端工具（如MQTTX）连接到服务端localhost:1883。
订阅和发布消息，观察服务端输出日志以确认服务端是否正确接收到消息。

步骤4：提升安全性

在生产环境中，为了提升MQTT服务端的安全性，可以通过设置用户名和密码来限制访问。以下是增强安全性的示例代码，使用了Aedes的authenticate方法来添加用户认证。

const aedes = require("aedes")();
const server = require("net").createServer(aedes.handle);
const PORT = 1883;// 用户认证信息（示例），包含用户名、密码和ClientID
const USERS = {user1: { password: "password1", clientId: "client1" },user2: { password: "password2", clientId: "client2" },
};server.listen(PORT, function () {console.log(`MQTT server is running on port ${PORT}`);
});// 用户认证方法，验证ClientID、用户名和密码
aedes.authenticate = (client, username, password, callback) => {const user = USERS[username];if (!user) {// 用户名不存在const error = new Error("Authentication Failed: Username not found");error.returnCode = 4;return callback(error, null);}if (user.password !== password.toString()) {// 密码错误const error = new Error("Authentication Failed: Incorrect password");error.returnCode = 4;return callback(error, null);}if (user.clientId !== client.id) {// ClientID不匹配const error = new Error(`Authentication Failed: ClientID "${client.id}" is not authorized`);error.returnCode = 4;return callback(error, null);}// 认证成功callback(null, true);
};aedes.on("client", (client) => {console.log(`Client Connected: ${client.id}`);
});aedes.on("clientDisconnect", (client) => {console.log(`Client Disconnected: ${client.id}`);
});aedes.on("publish", (packet, client) => {console.log(`Message Published: ${packet.payload.toString()}`);
});

步骤5：使用 Topic

在MQTT服务中，Topic（主题）是一个用于分类和过滤消息的机制。每个客户端可以订阅或发布特定的主题，服务端会根据主题将消息路由到相应的订阅者。

以下是一个示例，用于展示如何在Node.js中使用MQTT主题来管理和分类不同类型的消息。

场景：设备每 5 分钟上报一次温度，当服务端接收到阀值温度时，打开设备通风开关。

服务端核心代码

...const TEMPERATURE_THRESHOLD = 25;  // 温度阈值...
// 当接收到温度数据时，根据温度值决定是否开启设备
aedes.on('publish', (packet, client) => {const topic = packet.topic;const message = packet.payload.toString();if (topic === 'sensor/temperature') {console.log(`Received temperature: ${message}°C`);// 判断温度是否超过阈值const temperature = parseFloat(message);if (temperature > TEMPERATURE_THRESHOLD) {console.log('Temperature exceeds threshold. Sending switch ON command.');aedes.publish({ topic: 'command/switch', payload: 'ON' });}}
});