物联网服务NodeJs-5天学习第三天实战篇③ ——基于MQTT的环境温度检测

Posted 2023-03-13 单片机菜鸟哥

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了物联网服务NodeJs-5天学习第三天实战篇③ ——基于MQTT的环境温度检测相关的知识，希望对你有一定的参考价值。

【NodeJs-5天学习】第三天实战篇③ ——基于MQTT的环境温度检测

面向读者群体

❤️ 电子物联网专业同学，想针对硬件功能构造简单的服务器，不需要学习专业的服务器开发知识 ❤️
❤️ 业余爱好物联网开发者，有简单技术基础，想针对硬件功能构造简单的服务器❤️
❤️ 本篇创建记录 2023-03-12 ❤️
❤️ 本篇更新记录 2023-03-12 ❤️

技术要求

有HTML、CSS、JavaScript基础更好，当然也没事，就直接运行实例代码学习

专栏介绍

通过简短5天时间的渐进式学习NodeJs，可以了解到基本的服务开发概念，同时可以学习到npm、内置核心API（FS文件系统操作、HTTP服务器、Express框架等等），最终能够完成基本的物联网web开发，而且能够部署到公网访问。

🙏 此博客均由博主单独编写，不存在任何商业团队运营，如发现错误，请留言轰炸哦！及时修正！感谢支持！🎉 欢迎关注 🔎点赞 👍收藏 ⭐️留言📝

1. 前言

说到物联网，基本上离不开一个网络协议——MQTT。而在NodeJs中集成MQTT服务器也是非常简单易行，这里我们就构建一个简单的基于本地MQTT服务器的环境温度检测小系统。

2.实现思路

① 本地部署一个MQTT服务器，端口号是 1883，负责监听ESP8266通过mqtt协议上传上来的温度数据
② 本地部署一个Express服务器，端口号是8266，负责处理浏览器请求静态UI数据展示页面，静态页面由html、css、js编写
③ 核心业务处理包括处理温度数据，记录数据到fs文件系统
④ 设备端存在多个ESP8266, 每个8266都是一个Node节点的概念，上传该节点对应的ds18b20数据，我们需要对ds18b20进行编号，类似于#1，#2，#3等等。

2.1 NodeJs服务器代码

服务器代码包括两部分：

mqtt服务器，负责和设备端进行通信
express服务器，负责web页面展示数据

2.2.1 本地部署MQTT服务器，端口1883

const mqttTopic = require('./router/topic_router.js')
const getIPAdress = require('./utils/utils.js')
const deviceConfig = require('./config/device_config.js')
// 2、创建web服务器               
const myHost = getIPAdress();
const aedes = require('aedes')()
const server = require('net').createServer(aedes.handle)
const port = 1883
 
server.listen(port, function () 
  console.log("mqtt 服务器启动成功 mqtt://"+ myHost +":" + port);
);
 
// API文档：https://github.com/moscajs/aedes/blob/main/docs/Aedes.md
/********************************客户端连接状态************************************************************/

//Some Use Cases:
// - Rate Limit / Throttle by client.conn.remoteAddress
// - Check aedes.connectedClient to limit maximum connections
// - IP blacklisting
// Any error will be raised in connectionError event.
aedes.preConnect = function(client, packet, callback) 
    // 这个时候还没有 client.id
    console.log('服务器收到客户端: \\x1b[31m' + (client ? client.conn.remoteAddress : client) + '\\x1b[0m preConnect');
    callback(null, true)


// aedes.preConnect = function(client, packet, callback) 
//     callback(new Error('connection error'), client.conn.remoteAddress !== '::1') 
// 

// 连接身份验证,这个方法在preConnect之后
aedes.authenticate = function (client, username, password, callback) 
    console.log('客户端连接身份验证: \\x1b[31m' + (client ? client.id : client) + '\\x1b[0m authenticate');

    if (client.id && deviceConfig.authID(client.id)) 
       callback(null, deviceConfig.authLogin(client.id, username, password.toString()))
       return
    
    var error = new Error('Auth error,非法ID')
    error.returnCode = 4
    callback(error, null)


// aedes.authenticate = function (client, username, password, callback) 
//     var error = new Error('Auth error')
//     error.returnCode = 4
//     callback(error, null)
// 

// 客户端正在连接
aedes.on('client', function (client) 
    console.log('\\x1b[33m' + (client ? client.id : client) + '\\x1b[0m', '客户端正在连接到 broker', aedes.id);
);

// 客户端连接成功
aedes.on('clientReady', function (client) 
    console.log('\\x1b[33m' + (client ? client.id : client) + '\\x1b[0m', '客户端连接成功到 broker', aedes.id);
);
 
 
// 客户端连接断开
aedes.on('clientDisconnect', function (client) 
    console.log('\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '客户端连接断开 clientDisconnect');
);

// 客户端连接错误
aedes.on('clientError', function (client, error) 
    console.log('\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '客户端连接错误 clientError');
);


// 客户端连接异常
// Emitted when an error occurs. Unlike clientError it raises only when client is uninitialized.
aedes.on('connectionError', function (client, error) 
    console.log('\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '客户端连接异常 connectionError');
);

// CONNACK —— 确认连接请求
// （服务端发送CONNACK报文响应从客户端收到的CONNECT报文。服务端发送给客户端的第一个报文必须是CONNACK）
aedes.on('connackSent', function (packet , client ) 
    console.log('服务端确认连接给到\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m connackSent', packet);
);
/********************************客户端连接状态************************************************************/

/********************************心跳应答************************************************************/
// 客户端连接超时
aedes.on('keepaliveTimeout', function (client, error) 
    console.log('\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '客户端心跳连接超时 keepaliveTimeout');
);

// Emitted an QoS 1 or 2 acknowledgement when the packet successfully delivered to the client.
aedes.on('ack', function (packet , client) 
    console.log('服务端应答客户端: \\x1b[31m' + (client ? client.id : client) + '\\x1b[0m 内容：', packet);
);

// Emitted when client sends a PINGREQ.
aedes.on('ping', function (packet , client) 
    console.log('\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '客户端发送过来心跳 Ping');
);
/********************************心跳应答************************************************************/

/********************************主题相关************************************************************/

aedes.on('publish', function (packet, client) 
    if(client) 
        console.log('服务器收到客户端\\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '发布内容：',  packet);
        mqttTopic.route(packet.topic, packet.payload)
     else 
        console.log('服务器发布内容：',  packet);
    
);

// Emitted when client successfully subscribe the subscriptions in server.
aedes.on('subscribe', function (subscriptions , client) 
    console.log('服务器收到客户端: \\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '主题订阅：',  subscriptions);
);

// 对订阅主题进行校验
aedes.authorizeSubscribe = function (client, sub, callback) 
    // if (sub.topic === 'aaaa') 
    //   return callback(new Error('wrong topic'))
    // 
    // if (sub.topic === 'bbb') 
    //   // overwrites subscription
    //   sub.topic = 'foo'
    //   sub.qos = 1
    // 
    callback(null, sub)


// Emitted when client successfully unsubscribe the subscriptions in server.
aedes.on('unsubscribe', function (unsubscriptions , client) 
    console.log('服务器收到客户端: \\x1b[31m' + (client ? client.id : client) + '\\x1b[0m', '主题注销：',  unsubscriptions);
);
/********************************主题相关************************************************************/

2.2.1.1 用户校验


// 连接身份验证,这个方法在preConnect之后
aedes.authenticate = function (client, username, password, callback) 
    console.log('客户端连接身份验证: \\x1b[31m' + (client ? client.id : client) + '\\x1b[0m authenticate');

    if (client.id && deviceConfig.authID(client.id)) 
       callback(null, deviceConfig.authLogin(client.id, username, password.toString()))
       return
    
    var error = new Error('Auth error,非法ID')
    error.returnCode = 4
    callback(error, null)

这里会对clientid进行校验，通过之后再进行校验用户名字和密码。相当于三元组都得通过。

校验配置：

[
	"id": "1",
	"mac": "B0:E1:7E:70:25:CD",
    "name": "user",
    "psw": "123456"
, 
	"id": "2",
	"mac": "78:DA:07:04:5D:18",
    "name": "user",
    "psw": "123456"
, 
	"id": "3",
	"mac": "30:FC:68:19:52:A4",
    "name": "user",
    "psw": "123456"
]

包括四个元素：

clientid
mac地址
用户名字name
用户密码psw

接下来看看校验代码实现：

const fs = require('fs')

// 设备配置信息，需要通过配置校验才会通过
var fileName = './config/device.json';
var config = JSON.parse(fs.readFileSync(fileName));
var idToMacMap = new Map()
var idToValueMap = new Map()
config.forEach(element => 
  var id = element.id
  var mac = element.mac
  if (idToMacMap.has(id)) 
    throw console.error('device_config配置文件出现重复ID，请检查！' + id);
  
  idToMacMap.set(id, mac)
  idToValueMap.set(id, element)
)

console.log("用户设备配置信息：")
console.log(idToValueMap)

// 判断设备是否是合法设备，这里只是判断名字，最好是连着mac地址一起判断
function authID(deviceId, mac) 
    return idToMacMap.has(deviceId)


function authLogin(deviceId, name, psw) 
    var value = idToValueMap.get(deviceId)
    console.log(value)
    if (!value)
        console.log('不存在ID:' + deviceId)
        return false
    

    return value.name === name && value.psw === psw 


module.exports = 
    authID,
    authLogin

主要是把配置文件变成一个map映射对象，key是clientid。判断规则主要是判断是否存在对应的clientid，再把具体值拿出来进行下一轮比较。

2.2.1.2 主题消息处理

// 导入所需插件模块
const fs = require('fs')
const querystring = require('querystring')
const ds18b20Handler = require('./ds18b20_handler')

var mqttTopic = 
var routes = []

const USE_DEFAULT = 0
const USE_JSON = 1
const USE_FORM = 2

// 注入主题和处理方法
mqttTopic.use = function(path, dataType, action)
    if (!dataType) dataType = USE_DEFAULT
    routes.push([path, dataType, action])
 

// 路由匹配
mqttTopic.route = function(topic, payload) 
    for (let index = 0; index < routes.length; index++) 
      const route = routes[index];
      var key = route[0]
      var dataType = route[1]
      var action = route[2]
      if (topic === key) 
        var rawBody = payload.toString()
        if (dataType == USE_JSON) 
            try 
                action(JSON.parse(rawBody))
             catch (e) 
               // 异常内容
               console.log('Invalid JSON')
            
         else if(dataType == USE_FORM) 
            action(querystring.parse(req.rawBody))
         else 
            action(payload)
        
      
    


mqttTopic.use('SysThingPropertyPost',USE_JSON, (payload)=>
    console.log(payload)
    if (payload.params.temp) 
        ds18b20Handler.setCurrentTemp(payload.id, payload.params.temp)
        ds18b20Handler.saveToFile(payload.id, payload.params.temp)
    
 )

module.exports = 
    mqttTopic

主要是处理映射对应的mqtt主题，目前这里是处理 SysThingPropertyPost，这里是8266订阅的主题。

2.2.2 本地部署Express服务器，端口8266

// 1、导入所需插件模块
const express = require("express");
const getIPAdress = require('./utils/utils.js')
const bodyParser = require('body-parser')
const router = require('./router/router.js')

// 2、创建web服务器
let app = express();    
const port = 8266; // 端口号                 
const myHost = getIPAdress();

// 3、注册中间件，app.use 函数用于注册中间件
// 3.1 预处理中间件
app.use(bodyParser.json());
app.use(bodyParser.urlencoded( extended: true ));

app.use(function(req, res, next)
    // url地址栏出现中文则浏览器会进行iso8859-1编码，解决方案使用decode解码
    console.log('解码之后' + decodeURI(req.url));
    console.log('URL:' + req.url);
    if (req.method.toLowerCase() === 'post') 
        console.log(req.body);
    
    next()
)

// 3.2 路由中间件
app.use(router)
app.use(express.static('web')) // express.static()方法，快速对外提供静态资源

// 3.3 错误级别中间件(专门用于捕获整个项目发生的异常错误，防止项目奔溃),必须注册在所有路由之后
app.use((err, req, res, next) => 
    console.log('出现异常：' + err.message)
    res.send('Error: 服务器异常，请耐心等待！')
)

// 4、启动web服务器
app.listen(port,() => 
    console.log("express 服务器启动成功 http://"+ myHost +":" + port);
);

这里注入了api路由和静态文件路由。

// 3.2 路由中间件
app.use(router)
app.use(express.static('web')) // express.static()方法，快速对外提供静态资源

2.2.2.1 api路由中间件

// 1、导入所需插件模块
const express = require("express")
const ds18b20Handler = require('./ds18b20_handler.js')

// 2、创建路由对象
const router = express.Router();    

// 3、挂载具体的路由
// 配置add URL请求处理
物联网服务NodeJs-5天学习第三天实战篇④ ——QQ机器人，实现自动回复重要提醒
 物联网服务NodeJs-5天学习第三天实战篇① ——10行代码给她造个熬夜提醒睡觉机器人
 物联网服务NodeJs-5天学习第二天篇③ ——Express Web框架 和 中间件
 物联网服务NodeJs-5天学习第四天存储篇③ ——基于物联网的WiFi自动打卡考勤系统，升级存储为mysql，提醒功能改为QQ
 物联网服务NodeJs-5天学习第一天篇③ —— VsCode上运行第一个NodeJs 程序，配置自动重启插件 nodemon
 基于物联网的NodeJs-5天学习入门指引