本文详细讲解了基于合宙Air780EP模组LuatOS开发的多个MQTT应用示例。
本文同样适用于合宙的以下型号：
Air780EPA/Air780EPT/Air780EPS
Air780E/Air780EX/Air201…
一、相关准备工作
1.1 硬件准备
合宙EVB_Air780EP开发板一套，包括天线、SIM卡；
USB线
PC电脑
1.2 软件准备
登录合宙官方IoT后台：
https://iot.openluat.com/
LuatOS固件下载：
https://gitee.com/openLuat/LuatOS/releases
DEMO源码下载：
https://gitee.com/openLuat/LuatOS/tree/master/demo/mqtt
注：进入LuatOS固件页面按下Ctrl+F，搜索Air780EP即可找到对应的固件。
推荐选用：相关型号固件名称后面数字版本号最高的最新relase版本进行调试。
二、名词解析及API说明
2.1 名词解析
user_name用户名
password密码
client_id用户标识
MQTT可以通过前两个参数保证连接的安全，通过client_id确保设备唯一性。
QOS消息质量
分为0/1/2三个等级，分别表示只发一次，至少收到一次和只收到一次。通过QOS可以根据数据的重要性灵活，选择发送方式以节省带宽和保证数据可靠。
关于publish时QOS值的说明：
QOS为0时，压入底层TCP发送堆栈，视为成功；
QOS为1时，收到服务器回应PUBACK，视为成功；
QOS为2时，收到服务器响应PUBREC，立即上行PUBCOMP压入TCP发送队列，视为成功。
mqtt_host主机服务器地址
mqtt_port端口
MQTT需要连接的服务器参数。
mqtt_host可以说是ip或者域名。
topic主题
根据主题区别消息类型和来源，主要用来分类数据。同时mqtt是发布订阅模型，topic是发布和订阅者通信的重要通道。
payload消息内容
发布和订阅的具体数据。
retain保留消息
保留消息是一条将保留标志（retained flag）置为true的普通MQTT消息。broker会针对主题依照QoS级别保留最后一条保留消息，当订阅者订阅主题时会立即收到保留消息。broker仅为每个主题保留一条保留消息。
2.2 API说明
三、MQTT实现流程
3.1 创建一个TASK协程
接口：
sys.taskInit(func, arg1, arg2, argN)
参数：
返回值：
3.2 等待网络就绪
3.3 创建一个mqtt实例
接口：
mqtt.create(adapter,host,port,ssl,isipv6)
参数：
返回值：
3.4 连接服务器
接口：
mqttc:connect()
参数：
无
返回值：
注意：本函数仅代表发起成功, 后续仍需根据ready函数判断mqtt是否连接正常。
3.5 订阅主题
接口：
mqttc:subscribe(topic, qos)
参数：
返回值：
3.6 发布消息
接口：
mqttc:publish(topic, data, qos, retain)
参数：
返回值：
3.7 接收消息
3.8 断开服务器的连接
接口：mqttc:disconnect()
参数：无
返回值：
四、MQTT示例说明
本文以demo_lua\LuatOS\demo\mqtt这个demo为例作为演示，以MQTTX工具来测试MQTT运行状况。
**4.1 在MQTTX软件上建立连接 **
4.2 在MQTTX软件上添加订阅
注意：
订阅主题格式要求默认为 /luatos/pub/ 加模块的IMEI号。
例如：/luatos/pub/868488076506128
4.3 设置MQTTX软件上发布消息的主题
注意：
格式要求默认为 /luatos/sub/ 加模块的IMEI号。
例如：/luatos/sub/868488076506128
4.4 烧录脚本
上电开机后模块会自动向服务器发送消息，这是代码默认设置的，若不需要可以删掉。
4.5 模块向服务器发送消息
通过Luatools软件可以看到：
Luatools软件下载：
https://wiki.luatos.com/pages/tools.html
五、MQTT单链接示例
5.1 修改指定参数
在代码开头，可根据自己的服务器修改指定的参数。
注意：
user_name、password在有些服务器上是可以不传入的，或者是对传入的值没有要求限制；要根据实际服务器要求来填写。
--根据自己的服务器修改以下参数
local mqtt_host = "lbsmqtt.airm2m.com"
local mqtt_port = 1884
local mqtt_isssl = false
local client_id = "abc"
local user_name = "user"
local password = "password"
local pub_topic = "/luatos/pub/" .. (mcu.unique_id():toHex())
-- 该变量在下方代码有重新赋值，根据实际应用场景，可自行修改脚本逻辑
local sub_topic = "/luatos/sub/" .. (mcu.unique_id():toHex())
-- 该变量在下方代码有重新赋值，根据实际应用场景，可自行修改脚本逻辑
5.2 连接/订阅/发布流程
此task实现的是MQTT的连接、订阅消息、发布消息的流程。
注意：
要先等待网络就绪之后，才可进行MQTT后续操作。
待网络就绪之后，根据代码编写情况此时client_id、pub_topic和sub_topic会发生变化，会覆盖掉代码开头部分时的配置，这点需要注意；device_id为模块的IMEI号。
示例源码下载：
https://gitee.com/openLuat/LuatOS/tree/master/demo/mqtt
sys.taskInit(function()
-- 等待联网
local ret, device_id = sys.waitUntil("net_ready")
-- 下面的是mqtt的参数均可自行修改
-- client_id = device_id
pub_topic = "/luatos/pub/" .. device_id
sub_topic = "/luatos/sub/" .. device_id
-- 打印一下上报(pub)和下发(sub)的topic名称
-- 上报: 设备 ---> 服务器
-- 下发: 设备 <--- 服务器
-- 可使用mqtt.x等客户端进行调试
log.info("mqtt", "pub", pub_topic)
log.info("mqtt", "sub", sub_topic)
-- 打印一下支持的加密套件, 通常来说, 固件已包含常见的99%的加密套件
-- if crypto.cipher_suites then
-- log.info("cipher", "suites", json.encode(crypto.cipher_suites()))
-- end
if mqtt == nil then
while 1 do
sys.wait(1000)
log.info("bsp", "本bsp未适配mqtt库, 请查证")
end
end
-------------------------------------
-------- MQTT 演示代码 --------------
-------------------------------------
mqttc = mqtt.create(nil, mqtt_host, mqtt_port, mqtt_isssl, ca_file)
mqttc:auth(client_id,user_name,password) -- client_id必填,其余选填
-- mqttc:keepalive(240) -- 默认值240s
mqttc:autoreconn(true, 3000) -- 自动重连机制
mqttc:on(function(mqtt_client, event, data, payload)
-- 用户自定义代码
log.info("mqtt", "event", event, mqtt_client, data, payload)
if event == "conack" then
-- 联上了
sys.publish("mqtt_conack")
mqtt_client:subscribe(sub_topic)--单主题订阅
-- mqtt_client:subscribe({[topic1]=1,[topic2]=1,[topic3]=1})--多主题订阅
elseif event == "recv" then
log.info("mqtt", "downlink", "topic", data, "payload", payload)
sys.publish("mqtt_payload", data, payload)
elseif event == "sent" then
-- log.info("mqtt", "sent", "pkgid", data)
-- elseif event == "disconnect" then
-- 非自动重连时,按需重启mqttc
-- mqtt_client:connect()
end
end)
-- mqttc自动处理重连, 除非自行关闭
mqttc:connect()
sys.waitUntil("mqtt_conack")
while true do
-- 演示等待其他task发送过来的上报信息
local ret, topic, data, qos = sys.waitUntil("mqtt_pub", 300000)
if ret then
-- 提供关闭本while循环的途径, 不需要可以注释掉
if topic == "close" then break end
mqttc:publish(topic, data, qos)
end
-- 如果没有其他task上报, 可以写个空等待
--sys.wait(60000000)
end
mqttc:close()
mqttc = nil
end)
5.3 发送数据
此task的功能为模块每3秒向服务器发送一次数据。
-- 这里演示在另一个task里上报数据, 会定时上报数据,不需要就注释掉
sys.taskInit(function()
sys.wait(3000)
local data = "123,"
local qos = 1 -- QOS0不带puback, QOS1是带puback的
while true do
sys.wait(3000)
if mqttc and mqttc:ready() then
local pkgid = mqttc:publish(pub_topic, data .. os.date(), qos)
-- local pkgid = mqttc:publish(topic2, data, qos)
-- local pkgid = mqttc:publish(topic3, data, qos)
end
end
end)
5.4 mqtt-uart透传
此代码可实现mqtt-uart透传，利用串口工具给服务器发消息或者接收来着服务器的消息。
注意：
要使用串口1，且波特率为9600。
-- 以下是演示与uart结合, 简单的mqtt-uart透传实现,不需要就注释掉
local uart_id = 1
uart.setup(uart_id, 9600)
uart.on(uart_id, "receive", function(id, len)
local data = ""
while 1 do
local tmp = uart.read(uart_id)
if not tmp or #tmp == 0 then
break
end
data = data .. tmp
end
log.info("uart", "uart收到数据长度", #data)
sys.publish("mqtt_pub", pub_topic, data)
end)
sys.subscribe("mqtt_payload", function(topic, payload)
log.info("uart", "uart发送数据长度", #payload)
uart.write(1, payload)
end)
六、MQTT多链接示例
多链接的核心，就是要创建两个mqtt客户端的对象，通过不同的变量控制，代码部分如下。
示例源码下载：
https://gitee.com/openLuat/LuatOS/tree/master/demo/mqtt
--------------------第一个mqtt客户端--------------------
mqttc1 = mqtt.create(nil, mqtt_host, mqtt_port, mqtt_isssl, ca_file) -- 创建的第一个mqtt对象
mqttc1:auth(client1_id,user_name,password) -- client_id必填,其余选填
-- mqttc1:keepalive(240) -- 默认值240s
mqttc1:autoreconn(true, 3000) -- 自动重连机制
mqttc1:on(function(mqtt_client, event, data, payload)
-- 用户自定义代码
log.info("mqtt", "event", event, mqtt_client, data, payload)
if event == "conack" then
-- 联上了
sys.publish("mqtt_conack")
mqtt_client:subscribe(sub_topic_client)--单主题订阅
-- mqtt_client:subscribe({[topic1]=1,[topic2]=1,[topic3]=1})--多主题订阅
elseif event == "recv" then
-- 客户端1 接收数据
log.info("mqtt", "downlink", "topic", data, "payload", payload)
sys.publish("mqtt_payload", data, payload)
elseif event == "sent" then
-- log.info("mqtt", "sent", "pkgid", data)
-- elseif event == "disconnect" then
-- 非自动重连时,按需重启mqttc
-- mqtt_client:connect()
end~
end)
--------------------第二个mqtt客户端--------------------
mqttc2 = mqtt.create(nil, mqtt_host, mqtt_port, mqtt_isssl, ca_file) -- 创建的第二个mqtt对象
mqttc2:auth(client2_id,user_name,password) -- client_id必填,其余选填
-- mqttc2:keepalive(240) -- 默认值240s
mqttc2:autoreconn(true, 3000) -- 自动重连机制
mqttc2:on(function(mqtt_client, event, data, payload)
-- 用户自定义代码
log.info("mqtt", "event", event, mqtt_client, data, payload)
if event == "conack" then
-- 联上了
sys.publish("mqtt_conack")
mqtt_client:subscribe(sub_topic_client) -- 主题订阅 -> 订阅主题可以额外自定义
-- mqtt_client:subscribe({[topic1]=1,[topic2]=1,[topic3]=1})--多主题订阅
elseif event == "recv" then
-- 客户端2 接收数据
log.info("mqtt", "downlink", "topic", data, "payload", payload)
sys.publish("mqtt_payload", data, payload)
elseif event == "sent" then
-- log.info("mqtt", "sent", "pkgid", data)
-- elseif event == "disconnect" then
-- 非自动重连时,按需重启mqttc
-- mqtt_client:connect()
end
end)
-- 客户端1 发送数据
mqttc1:publish(topic, data, qos) -- 发布topic主题可以自定义，可以不相同
-- 客户端2 发送数据
mqttc2:publish(topic, data, qos) -- 发布topic主题可以自定义，可以不相同
七、MQTT SSL不带证书连接示例
示例源码下载：
https://gitee.com/openLuat/LuatOS/tree/master/demo/mqtt
local mqtt_host = "broker.emqx.io"
local mqtt_port = 8883
local mqtt_isssl = true -- 是否使用ssl> false 不加密 | true 无证书加密 | table 有证书加密
local client_id = "abc"
local user_name = "user"
local password = "password"
local pub_topic = "/luatos/pub/"
local sub_topic = "/luatos/sub/"
-- 统一联网函数
sys.taskInit(function()
local device_id = mcu.unique_id():toHex()
-----------------------------
-- 统一联网函数, 可自行删减
----------------------------
if wlan and wlan.connect then
-- wifi 联网, ESP32系列均支持
local ssid = "luatos1234"
local password = "12341234"
log.info("wifi", ssid, password)
-- TODO 改成自动配网
-- LED = gpio.setup(12, 0, gpio.PULLUP)
wlan.init()
wlan.setMode(wlan.STATION) -- 默认也是这个模式,不调用也可以
device_id = wlan.getMac()
wlan.connect(ssid, password, 1)
elseif mobile then
-- Air780E/Air600E系列
--mobile.simid(2) -- 自动切换SIM卡
-- LED = gpio.setup(27, 0, gpio.PULLUP)
device_id = mobile.imei()
elseif w5500 then
-- w5500 以太网, 当前仅Air105支持
w5500.init(spi.HSPI_0, 24000000, pin.PC14, pin.PC01, pin.PC00)
w5500.config() --默认是DHCP模式
w5500.bind(socket.ETH0)
-- LED = gpio.setup(62, 0, gpio.PULLUP)
elseif socket or mqtt then
-- 适配的socket库也OK
-- 没有其他操作, 单纯给个注释说明
else
-- 其他不认识的bsp, 循环提示一下吧
while 1 do
sys.wait(1000)
log.info("bsp", "本bsp可能未适配网络层, 请查证")
end
end
-- 默认都等到联网成功
sys.waitUntil("IP_READY")
sys.publish("net_ready", device_id)
end)
sys.taskInit(function()
-- 等待联网
local ret, device_id = sys.waitUntil("net_ready")
-- 下面的是mqtt的参数均可自行修改
client_id = device_id
pub_topic = "/luatos/pub/" .. device_id
sub_topic = "/luatos/sub/" .. device_id
-- 打印一下上报(pub)和下发(sub)的topic名称
-- 上报: 设备 ---> 服务器
-- 下发: 设备 <--- 服务器
-- 可使用mqtt.x等客户端进行调试
log.info("mqtt", "pub", pub_topic)
log.info("mqtt", "sub", sub_topic)
if mqtt == nil then
while 1 do
sys.wait(1000)
log.info("bsp", "本bsp未适配mqtt库, 请查证")
end
end
-------------------------------------
-------- MQTT 演示代码 --------------
-------------------------------------
mqttc = mqtt.create(nil, mqtt_host, mqtt_port, mqtt_isssl, ca_file)
mqttc:auth(client_id,user_name,password) -- client_id必填,其余选填
-- mqttc:keepalive(240) -- 默认值240s
mqttc:autoreconn(true, 3000) -- 自动重连机制
mqttc:on(function(mqtt_client, event, data, payload)
-- 用户自定义代码
log.info("mqtt", "event", event, mqtt_client, data, payload)
if event == "conack" then
-- 联上了
sys.publish("mqtt_conack")
mqtt_client:subscribe(sub_topic)--单主题订阅
-- mqtt_client:subscribe({[topic1]=1,[topic2]=1,[topic3]=1})--多主题订阅
elseif event == "recv" then
log.info("mqtt", "downlink", "topic", data, "payload", payload)
sys.publish("mqtt_payload", data, payload)
elseif event == "sent" then
-- log.info("mqtt", "sent", "pkgid", data)
-- elseif event == "disconnect" then
-- 非自动重连时,按需重启mqttc
-- mqtt_client:connect()
end
end)
-- mqttc自动处理重连, 除非自行关闭
mqttc:connect()
sys.waitUntil("mqtt_conack")
while true do
-- 演示等待其他task发送过来的上报信息
local ret, topic, data, qos = sys.waitUntil("mqtt_pub", 300000)
if ret then
-- 提供关闭本while循环的途径, 不需要可以注释掉
if topic == "close" then break end
mqttc:publish(topic, data, qos)
end
-- 如果没有其他task上报, 可以写个空等待
--sys.wait(60000000)
end
mqttc:close()
mqttc = nil
end)
八、MQTT SSL带证书连接示例
示例源码下载：
https://gitee.com/openLuat/LuatOS/tree/master/demo/mqtt
local mqtt_host = "broker.emqx.io"
local mqtt_port = 8883
local mqtt_isssl = { -- 是否使用ssl> false 不加密 | true 无证书加密 | table 有证书加密
server_cert = io.readFile("/luadb/broker.emqx.io-ca.crt"), -- 把证书文件作为脚本文件一起烧录到模块内，就可以用/luadb/路径直接读取
client_cert=nil,
client_key=nil,
client_password=nil,
}
local client_id = "abc"
local user_name = "user"
local password = "password"
local pub_topic = "/luatos/pub/"
local sub_topic = "/luatos/sub/"
-- 统一联网函数
sys.taskInit(function()
local device_id = mcu.unique_id():toHex()
-----------------------------
-- 统一联网函数, 可自行删减
----------------------------
if wlan and wlan.connect then
-- wifi 联网, ESP32系列均支持
local ssid = "luatos1234"
local password = "12341234"
log.info("wifi", ssid, password)
-- TODO 改成自动配网
-- LED = gpio.setup(12, 0, gpio.PULLUP)
wlan.init()
wlan.setMode(wlan.STATION) -- 默认也是这个模式,不调用也可以
device_id = wlan.getMac()
wlan.connect(ssid, password, 1)
elseif mobile then
-- Air780E/Air600E系列
--mobile.simid(2) -- 自动切换SIM卡
-- LED = gpio.setup(27, 0, gpio.PULLUP)
device_id = mobile.imei()
elseif w5500 then
-- w5500 以太网, 当前仅Air105支持
w5500.init(spi.HSPI_0, 24000000, pin.PC14, pin.PC01, pin.PC00)
w5500.config() --默认是DHCP模式
w5500.bind(socket.ETH0)
-- LED = gpio.setup(62, 0, gpio.PULLUP)
elseif socket or mqtt then
-- 适配的socket库也OK
-- 没有其他操作, 单纯给个注释说明
else
-- 其他不认识的bsp, 循环提示一下吧
while 1 do
sys.wait(1000)
log.info("bsp", "本bsp可能未适配网络层, 请查证")
end
end
-- 默认都等到联网成功
sys.waitUntil("IP_READY")
sys.publish("net_ready", device_id)
end)
sys.taskInit(function()
-- 等待联网
local ret, device_id = sys.waitUntil("net_ready")
-- 下面的是mqtt的参数均可自行修改
client_id = device_id
pub_topic = "/luatos/pub/" .. device_id
sub_topic = "/luatos/sub/" .. device_id
-- 打印一下上报(pub)和下发(sub)的topic名称
-- 上报: 设备 ---> 服务器
-- 下发: 设备 <--- 服务器
-- 可使用mqtt.x等客户端进行调试
log.info("mqtt", "pub", pub_topic)
log.info("mqtt", "sub", sub_topic)
if mqtt == nil then
while 1 do
sys.wait(1000)
log.info("bsp", "本bsp未适配mqtt库, 请查证")
end
end
-------------------------------------
-------- MQTT 演示代码 --------------
-------------------------------------
mqttc = mqtt.create(nil, mqtt_host, mqtt_port, mqtt_isssl, ca_file)
mqttc:auth(client_id,user_name,password) -- client_id必填,其余选填
-- mqttc:keepalive(240) -- 默认值240s
mqttc:autoreconn(true, 3000) -- 自动重连机制
mqttc:on(function(mqtt_client, event, data, payload)
-- 用户自定义代码
log.info("mqtt", "event", event, mqtt_client, data, payload)
if event == "conack" then
-- 联上了
sys.publish("mqtt_conack")
mqtt_client:subscribe(sub_topic)--单主题订阅
-- mqtt_client:subscribe({[topic1]=1,[topic2]=1,[topic3]=1})--多主题订阅
elseif event == "recv" then
log.info("mqtt", "downlink", "topic", data, "payload", payload)
sys.publish("mqtt_payload", data, payload)
elseif event == "sent" then
-- log.info("mqtt", "sent", "pkgid", data)
-- elseif event == "disconnect" then
-- 非自动重连时,按需重启mqttc
-- mqtt_client:connect()
end
end)
-- mqttc自动处理重连, 除非自行关闭
mqttc:connect()
sys.waitUntil("mqtt_conack")
while true do
-- 演示等待其他task发送过来的上报信息
local ret, topic, data, qos = sys.waitUntil("mqtt_pub", 300000)
if ret then
-- 提供关闭本while循环的途径, 不需要可以注释掉
if topic == "close" then break end
mqttc:publish(topic, data, qos)
end
-- 如果没有其他task上报, 可以写个空等待
--sys.wait(60000000)
end
mqttc:close()
mqttc = nil
end)
九、常见问题Q&A
以下针对客户朋友们实际应用中的反馈，整理了MQTT应用中的常见问题：
-
模组支持MQTT最新的版本是多少？
支持MQTT_V3.1、MQTT_V3.1.1版本。 -
模组最多支持几路链接？
MQTT/TCP/UDP的链接公用8路通道。 -
client id是否允许重复？
client id不允许重复，要保证唯一性，如果在已经有一个该id的链接情况下，另外一个设备也用了相同的clientid接入，会把另一个在连链接挤掉。 -
Qos0、Qos1、Qos2如何选择？
应用允许的情况下，建议使用Qos0；Qos1和Qos2会加重网络负担，4G网络还好，特别是2G网络，在网络拥堵和较差的情况下，数据传输的次数越多，掉线的概率就会越高。 -
为什么mqtt ssl访问失败？
检查服务器是否支持模块支持的加密套件，用模块程序中的三元组和证书，在mqttx或者mqttfx客户端工具上尝试是否可以成功链接。 -
排查是否设备单体问题：
如果同一地点，某些设备正常，某些设备异常，按照如下几种情况分析。
01）分析正常设备和异常设备的使用环境是否相同：
如果不同，例如异常设备固定在钢制墙壁上，正常设备放置在桌子上，钢制墙壁可能对天线射频有干扰，将异常设备和正常设备放置在同样的使用环境中，再对比测试。
如果相同，参考下方第2）步。
2）分析正常和异常的设备，驻留的小区是否相同：
如果相同，重点排查异常设备的天线射频部分。
如同不同，多测试几次，确认下，是不是在异常小区内很容易出问题，如果异常小区很容易出问题，可能就是小区拥堵造成的。