【Laf 物联网探索】Laf 使用MQTT 的正确姿势以及基础流程

wicos · 2023年6月16日

一、阅前须知

了解本文您需要了解的编程语言有：typescript
~~在进阶阅读部分您还需要掌握：硬件IO以及总线的操作~~
本文不涉及MQTT Broker 的搭建，仅从客户端的角度分析、操作

二、这是什么

本文主要讲解关于Laf与MQTT的实践结合以及分析思路。

2.1 为什么用MQTT而不是Websockers?

详见三中MQTT的优势，以及MQTT协议更契合传统硬件开发，同时也方便与其他人合作。

三、什么是MQTT

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。
MQTT最大优点在于，用极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。详细信息点我

四、思路分析

4.1 基础思路分析

因为Laf内的代码的运行环境为node，因此在开发MQTT时首先想到的是MQTT的包，那么我们只需要在Laf内引入mqtt包即可，在mqtt建立连接后，可以进行订阅、发送、接收等操作。

4.2 流程图思路梳理

五、伪代码展示

5.1 init 初始化代码

import { connect } from "mqtt"
import cloud from '@lafjs/cloud'

const mqttUrl = "mqtt://broker.emqx.io:1883"
const clientId = "laf_000001"
const mqttClient = connect(mqttUrl, {
    clientId,
    clean: true,
    connectTimeout: 4000,
    reconnectPeriod: 1000,
})

let mqttState = false
interface PUBCMD {
    topic: string,
    content: string | object,
    qos: number,
    retain: boolean
}

let topicArray = []

export default async function (ctx: FunctionContext) {
    cloud.shared.set("topicArray",topicArray)
    mqttClient.on("connect", () => {
        mqttState = true
        //此处为订阅函数
        mqttClient.subscribe(Array.from(new Set(topicArray)),()=>{
            console.log(`connected ${topicArray}`)
        })
    })
    //收到消息
    mqttClient.on("message",(topic,payload)=>{
        //收到消息的处理函数
        //...
    })
    cloud.shared.set("mqttClient", mqttClient)
    cloud.shared.set("mqttPubQueue", mqttPubQueue)



}

5.2 发送数据的函数 mqtt_pub

import cloud from '@lafjs/cloud'
const mqttClient = cloud.shared.get("mqttClient")
const mqttState = cloud.shared.get("mqttState")
export default async function (ctx: FunctionContext) {
    const { body } = ctx
    const { topic, content, qos = 0, retain = false } = body
    if (!topic || !content) {
        return
    }
    if (!mqttState) {
        await cloud.invoke("__init__")
    }
    let pubContent
    try {
        pubContent = JSON.stringify(content)
    } catch {
        pubContent = content.toString()
    }
    mqttClient.publish(topic, pubContent, { qos: qos, retain: retain })
}

5.3 接收函数

由于接收数据的函数难度不大，因此略过

六、稳定性分析

关于整体的稳定性，主要有几个方面，首先是mqtt broker的稳定性，这里不在讨论的范围，因为这里用的是EMQX公司的公共服务器（不建议用在生产环境），其次是Laf的稳定性，由于要在初始化函数内执行操作，目前来看问题不大。