物联网嵌入式【免费下载链接】nodemcu-firmwareLua based interactive firmware for ESP8266, ESP8285 and ESP32项目地址https://gitcode.com/gh_mirrors/no/nodemcu-firmware点击查看免费下载本指南以 NodeMCU 固件内置的mqtt模块为主线系统讲解在 ESP8266/ESP8285/ESP32 平台上如何用 Lua 语言创建 MQTT 客户端、建立安全连接、订阅发布消息、处理遗嘱LWT与大消息溢出等完整场景并结合 app/modules/mqtt.c 源码剖析消息缓冲、心跳保活与 QoS 应答的实现原理。读完本文你将能够编写一套具备自动重连、遗嘱上报、按主题分发能力的完整 MQTT 应用。模块概述与协议版本要求mqtt模块是 NodeMCU 固件内置的 MQTT 客户端实现源码位于 app/modules/mqtt.c底层报文编解码依赖 app/mqtt 目录下的mqtt_msg.c、msg_queue.c组件。重要前提该客户端遵循MQTT 协议 3.1.1 版本。在使用前请确认你的 Broker如 Mosquitto、EMQX、CloudMQTT 等支持并正确配置为 3.1.1 版本该客户端与运行 MQTT 3.1 旧协议的 Broker 不兼容后向不兼容否则会在连接阶段直接收到CONNACK_REFUSED_PROTOCOL_VER错误。模块 API 概览API作用mqtt.Client()创建 MQTT 客户端对象mqtt:connect()连接 Brokermqtt:close()主动断开连接发送 DISCONNECTmqtt:lwt()配置 Last Will and Testament遗嘱消息mqtt:on()注册事件回调mqtt:publish()发布消息mqtt:subscribe()/mqtt:unsubscribe()订阅 / 取消订阅创建客户端mqtt.Client()语法与参数mqtt.Client(clientid, keepalive[, username, password, cleansession, max_message_length])参数说明clientid客户端 ID字符串。可省略源码中若省略会自动生成为NodeMCU_ 芯片 ID 十六进制串见 app/modules/mqtt.ckeepalive心跳保活间隔秒。源码中若传 0 或省略默认回退到 60 秒MQTT_DEFAULT_KEEPALIVE见 app/modules/mqtt.cusername用户名可选password密码可选cleansession0/1对应false/true。默认 1true即每次连接都建立全新会话max_message_length可接收的最大消息长度默认 1024 字节返回一个 MQTT 客户端对象。max_message_length 与内存安全根据 MQTT 规范PUBLISH 报文的理论上限是 256MB这对 NodeMCU 的 RAM 来说显然不现实。为了避免内存耗尽OOM固件强制限制可接收消息的大小该限制即由max_message_length参数控制默认值 1024 字节是刻意选定的——这是 NodeMCU 2.2.1 及更早版本中的隐含限制当时完全没有显式处理。“消息长度”指完整的 MQTT 报文大小包含固定头、可变头、主题名、包 ID如适用和 Payload。精确定义请参阅 MQTT v3.1.1 规范。实践中该参数只影响接收方向的 PUBLISH报文因为所有常规控制报文都很小。任何大于max_message_length的消息其能收到的部分会被投递到overflow回调如果已注册其余部分被丢弃后续消息不受影响。即使消息被丢弃如果 QoS 为 1 或 2固件仍会向 Broker 发送 ACK——对应源码中mqtt_msg_puback/mqtt_msg_pubrec的入队逻辑见 app/modules/mqtt.c。跨 TCP 包的大消息缓冲机制一个容易被忽视的内存细节即使单条消息没有超过max_message_length也可能触发堆内存分配。原因如下当一条消息跨越多个 TCP 包时固件会用堆内存缓冲不完整部分。从源码看固件在首次看到消息头时就按完整消息长度一次性分配calloc避免反复 realloc 造成堆碎片见 app/modules/mqtt.c。如果分配失败MQTT 会话会被直接断开。Broker 可能在同一个 TCP 包中连续发送多条小消息若最后一条消息在包内装不下固件也会分配堆缓冲等待下一个 TCP 包。单条消息能塞进一个 TCP 包的典型上限是1460 字节与固件内部MQTT_BUF_SIZE一致见 app/modules/mqtt.c但实际取决于网络 MTU 配置、分包情况以及同一包内是否混有多条消息。接收状态机在源码中体现为tReceiveState的四种状态MQTT_RECV_NORMAL、MQTT_RECV_BUFFERING_SHORT、MQTT_RECV_BUFFERING、MQTT_RECV_SKIPPING见 app/modules/mqtt.c。其中SKIPPING状态专门用于“超过上限的消息”固件会丢弃后续 TCP 包中的剩余字节而不做存储。基础示例-- 无登录信息keepalive 120 秒 m mqtt.Client(clientid, 120) -- 带登录信息keepalive 120 秒 m mqtt.Client(clientid, 120, user, password) -- 配置遗嘱消息可选 -- 若客户端在 keepalive 时间内未发送心跳包Broker 会向 /lwt 发布一条 -- qos 0、retain 0、payload offline 的消息 m:lwt(/lwt, offline, 0, 0) -- 离线事件 m:on(offline, function(client) print(offline) end) -- 收到消息事件 m:on(message, function(client, topic, data) print(topic .. :) if data ~ nil then print(data) end end) -- 溢出事件收到被截断的超限消息 m:on(overflow, function(client, topic, data) print(topic .. partial overflowed message: .. data) end) -- TLS 连接写法m:connect(192.168.11.118, secure-port, 1) m:connect(192.168.11.118, 1883, false, function(client) print(connected) -- 订阅与发布必须在连接建立成功后才进行 -- 可以放在这里connect 回调或通过跟踪连接状态保证时机。 -- 订阅主题QoS 0 client:subscribe(/topic, 0, function(client) print(subscribe success) end) -- 发布消息data hello, QoS 0, retain 0 client:publish(/topic, hello, 0, 0, function(client) print(sent) end) end, function(client, reason) print(Connection failed reason: .. reason) end) m:close() -- offline 回调触发后可以再次调用 m:connect注册事件回调mqtt:on()mqtt:on(event, function(client[, topic[, message]]))回调函数的第一个参数永远是客户端对象本身其余参数因事件而异。事件回调参数说明connectclient连接建立成功connfailclient, reason连接失败reason为失败码见下表subackclient订阅成功应答unsubackclient取消订阅成功应答pubackclient发布应答QoS 1/2或发送完成QoS 0messageclient, topic, message收到 PUBLISH 消息topic/message均为 Lua 字符串overflowclient, topic, message收到超限消息message被截断到max_message_lengthofflineclient已建立连接被关闭注意与connfail不同见下文源码中的事件名白名单与回调注册逻辑可在 app/modules/mqtt.c 查看connfail对应的回调槽位是cb_connect_fail_ref。一个需要留意的事实suback/unsuback/puback等事件不携带额外参数这带来两个后果——Broker 返回的“订阅允许的最大 QoS”信息会丢失如果你期望逐事件确认就必须自行管理消息队列源码中的pending_msg_q只是内部控制包队列并不暴露给 Lua 层。连接 Brokermqtt:connect()语法与参数mqtt:connect(host[, port[, secure]][, function(client)[, function(client, reason)]])参数说明hostBroker 主机名、域名或 IP字符串portBroker 端口数字默认 1883见 app/modules/mqtt.csecure布尔值true表示启用 TLSmqtts。请留意 net 模块 中记录的约束function(client)连接建立成功回调function(client, reason)连接失败回调失败后不应再调用其他回调返回nil结果一律通过回调观察。attention安全连接mqtts带有不少限制请务必阅读 tls 模块 文档中的警告。回调别名规则关键语义:connect()的第一个回调与:on(connect, ...)互为别名二者中最后传入者生效但若向:connect()传入nil已有的回调会被保留而非清除。第二个失败回调与:on(connfail, ...)互为别名。offline回调只在“已建立”的连接变为关闭时触发。如果connect()本身就没建立成功只会调用:connect()传入的失败回调offline不会被触发。这一语义在源码连接失败路径中体现为直接调用mqtt_connack_fail见 app/modules/mqtt.c。旧文档曾建议用整数 0/1传secure现在这样写会触发弃用警告请改用布尔值false/true源码中通过platform_print_deprecation_note提示见 app/modules/mqtt.c。连接失败回调原因码固件通过mqtt:on(connfail, ...)的第二个参数返回失败原因码这些常量定义在 app/mqtt/mqtt_msg.h并在模块加载时以mqtt.CONN_*常量导出见 app/modules/mqtt.c常量值描述mqtt.CONN_FAIL_SERVER_NOT_FOUND-5指定 IP 与端口上没有 Broker 在监听mqtt.CONN_FAIL_NOT_A_CONNACK_MSG-4Broker 的响应不是协议要求的 CONNACK 报文mqtt.CONN_FAIL_DNS-3DNS 解析失败mqtt.CONN_FAIL_TIMEOUT_RECEIVING-2等待 Broker CONNACK 超时mqtt.CONN_FAIL_TIMEOUT_SENDING-1发送 CONNECT 报文超时mqtt.CONNACK_ACCEPTED0无错误。注意该值不会触发失败回调mqtt.CONNACK_REFUSED_PROTOCOL_VER1Broker 不是 MQTT 3.1.1 Brokermqtt.CONNACK_REFUSED_ID_REJECTED2Broker 拒绝了指定的 ClientID参见mqtt.Client()mqtt.CONNACK_REFUSED_SERVER_UNAVAILABLE3服务器不可用mqtt.CONNACK_REFUSED_BAD_USER_OR_PASS4Broker 拒绝了指定的用户名或密码mqtt.CONNACK_REFUSED_NOT_AUTHORIZED5用户名未授权这些原因码与源码中的处理路径一一对应例如等待 CONNACK 超时对应MQTT_CONN_FAIL_TIMEOUT_RECEIVING见 app/modules/mqtt.c收到的不是 CONNACK 报文对应MQTT_CONN_FAIL_NOT_A_CONNACK_MSG见 app/modules/mqtt.cCONNACK 返回码非 0 则直接透传对应拒绝码见 app/modules/mqtt.c。实现可靠的自动重连应用应当监听连接失败并在错误回调中处理才能实现可靠的服务器连接。官方推荐的重试模式function handle_mqtt_error(client, reason) tmr.create():alarm(10 * 1000, tmr.ALARM_SINGLE, do_mqtt_connect) end function do_mqtt_connect() mqtt:connect(server, function(client) print(connected) end, handle_mqtt_error) end当然真正的项目中“connected”回调里应该做点有用的事订阅、发布、状态上报等。连接过程的底层调用链从源码看mqtt:connect()的完整链路是解析域名dns_gethostbyname失败返回MQTT_CONN_FAIL_DNS见 app/modules/mqtt.c→ 通过espconn_connect或 TLS 下的espconn_secure_connect建立 TCP 连接见 app/modules/mqtt.c→ TCP 连接建立后组装 CONNECT 报文并立即发送mqtt_msg_connect见 app/modules/mqtt.c→ 进入MQTT_CONNECT_SENDING/MQTT_CONNECT_SENT状态等待 CONNACK。整个过程有MQTT_SEND_TIMEOUT5 秒超时保护。配置遗嘱消息mqtt:lwt()Last Will and Testament遗嘱消息是 MQTT 的重要机制客户端异常掉线时由 Broker 代为发布一条预设消息常用于设备离线通知。mqtt:lwt(topic, message[, qos[, retain]])参数说明topic遗嘱发布主题字符串message遗嘱消息内容buffer 或字符串qosQoS 等级默认 0retainretain 标志默认 0遗嘱在连接时发送给 Broker因此lwt()必须在connect()之前调用。源码中遗嘱信息保存在conf结构的will_topic_ref/will_message_ref/will_qos/will_retain字段中并在建立 TCP 连接后随 CONNECT 报文一起组装见 app/modules/mqtt.c。Broker 何时发布遗嘱当 Broker 发现与客户端的连接中断时就会发布遗嘱消息。触发条件包括客户端在mqtt.Client()指定的 keepalive 时间内没有发送心跳包TCP 连接被正常关闭但在关闭前没有先关闭 MQTT 连接Broker 尝试向客户端发送数据时 TCP 连接断裂。例如你指定 keepalive 为 120 秒那么直接关掉设备电源、且 Broker 期间没有向该客户端推送任何数据时遗嘱消息会在掉线约120 秒后被发布。已知限制注意当前 NodeMCU MQTT 库存在一个 bug所有断连都会被表现为“意外断连”——MQTT 层的断开消息未在 TCP 连接拆除前发出。结果就是LWT 遗嘱消息几乎总是会被发布。跟踪见 nodemcu-firmware issue #3031。发布消息mqtt:publish()mqtt:publish(topic, payload, qos, retain[, function(client)])参数说明topic发布主题字符串遵循 MQTT 主题规范如a/b/c、通配符主题用于订阅而非发布payload消息内容buffer 或字符串qosQoS 等级0/1/2retainretain 标志0/1function(client)可选回调QoS 1/2 时收到 PUBACK 后触发QoS 0 时消息发送后触发返回true表示成功false表示失败。注意事项多次调用publish()时最后一次定义的回调会被所有 publish 命令调用。该回调参数与:on(puback, ...)互为别名。QoS 1 的应答在源码中通过MQTT_MSG_TYPE_PUBACK状态处理QoS 2 走完整的 PUBREC → PUBREL → PUBCOMP 四步握手见 app/modules/mqtt.c。源码确认QoS 0 消息在 TCP 发送完成mqtt_socket_sent时即触发回调并出队因为不会收到服务端 PUBACK见 app/modules/mqtt.c。订阅与取消订阅mqtt:subscribe()mqtt:subscribe(topic, qos[, function(client)]) mqtt:subscribe(table[, function(client)])参数说明topic主题字符串qos订阅 QoS 等级默认 0tabletopic, qos键值对数组一次订阅多个主题function(client)可选回调订阅成功后触发-- 订阅单主题QoS 0 m:subscribe(/topic, 0, function(conn) print(subscribe success) end) -- 一次订阅多个主题topic/0 → qos0; topic/1 → qos1; topic2 → qos2 m:subscribe({[topic/0]0, [topic/1]1, topic22}, function(conn) print(subscribe success) end)caution如果需要订阅多个主题不要多次调用subscribe()请使用上例所示的多主题 table 语法一次性订阅。多次调用subscribe()时最后一次定义的回调会被所有 subscribe 命令调用并与:on(suback, ...)互为别名。mqtt:unsubscribe()mqtt:unsubscribe(topic[, function(client)]) mqtt:unsubscribe(table[, function(client)])参数说明topic主题字符串tabletopic, anything键值对数组一次取消多个订阅值可为任意值源码只读取键function(client)可选回调取消订阅成功后触发-- 取消订阅单主题 m:unsubscribe(/topic, function(conn) print(unsubscribe success) end) -- 一次取消多个主题值可为任意值如 0 或 anything m:unsubscribe({[topic/0]0, [topic/1]0, topic2anything}, function(conn) print(unsubscribe success) end)同样地多次调用时最后一次定义的回调会被所有 unsubscribe 命令调用并与:on(unsuback, ...)互为别名。源码实现细节单主题/多主题订阅都经由mqtt_msg_subscribe_*系列函数在栈上临时缓冲区组装成 SUBSCRIBE 报文再整体入队发送若主题过多导致缓冲区溢出会返回buffer overflow, cant enqueue all subscriptions错误见 app/modules/mqtt.c。主动断开mqtt:close()mqtt:close()参数无返回nilclose()调度一次干净的连接拆除源码中它会先构造 DISCONNECT 报文入队并发送见 app/modules/mqtt.c。MQTT 协议要求客户端主动向服务器发送断开意愿以避免触发遗嘱消息。因此调用close()后客户端不能立即复用必须等到offline回调触发后才能再次connect()。综合实战云端 MQTT 收发示例以下示例来自仓库 lua_examples/mqtt/mqtt2cloud.lua演示了连接云 Broker、按主题分发消息、周期性上报的完整模式-- test with cloudmqtt.com local m_dis {} -- 按主题分发收到的消息 local function dispatch(m, t, pl) if pl ~ nil and m_dis[t] then m_dist end end local function topic1func(_, pl) print(get1: .. pl) end local function topic2func(_, pl) print(get2: .. pl) end do m_dis[/topic1] topic1func m_dis[/topic2] topic2func -- 创建客户端clientid, keepalive60, user, pass local m mqtt.Client(nodemcu1, 60, test, test123) m:on(connect, function(client) print(connection .. node.heap()) client:subscribe(/topic1, 0, function() print(sub done) end) client:subscribe(/topic2, 0, function() print(sub done) end) client:publish(/topic1, hello, 0, 0) client:publish(/topic2, world, 0, 0) end) m:on(offline, function() print(disconnect to broker...) print(node.heap()) end) m:on(message, dispatch) -- 连接host, port, secure m:connect(m11.cloudmqtt.com, 11214, 0) -- 每 10 秒上报一次时间 tmr.create():alarm(10000, tmr.ALARM_AUTO, function() local pl time: .. tmr.time() m:publish(/topic1, pl, 0, 0) end) end底层实现要点回顾保活机制keepalive计时由软件定时器mqtt_socket_timer驱动见 app/modules/mqtt.c。进入MQTT_DATA状态后若定时器到期且发送队列为空则发送 PINGREQ 心跳若上一次 PINGREQ 已发出但未收到 PINGRESPkeepalive_sent仍为 1则判定心跳失联并直接断开连接。发送队列所有控制报文先经msg_enqueue进入pending_msg_q队列再由mqtt_send_if_possible逐条发送见 app/modules/mqtt.c保证与 Broker 的应答严格按序交互。接收处理mqtt_socket_received解析可变长消息头按tReceiveState状态机处理正常/缓冲/跳过三种情形并在一个 TCP 包内循环处理多条消息READPACKET循环见 app/modules/mqtt.c。模块导出mqtt.Client及全部CONN_*/CONNACK_*常量通过 LROT 表导出到 Lua 命名空间见 app/modules/mqtt.c这也是 Lua 层能直接引用mqtt.CONN_FAIL_DNS等常量的原因。小结NodeMCU 的mqtt模块为资源受限的 ESP 系列芯片提供了完整的 MQTT 3.1.1 客户端能力从连接、订阅发布、遗嘱配置到超限消息的溢出保护均有清晰的 Lua API 与源码级保障。开发时请特别关注三点Broker 必须是 MQTT 3.1.1、遗嘱必须在 connect 之前配置、close()后须等待offline回调才能重连。结合max_message_length的合理配置默认 1024可根据业务 Payload 大小上调但需评估堆内存占用即可构建稳定可靠的设备端消息通道。赞分享物联网嵌入式【免费下载链接】nodemcu-firmwareLua based interactive firmware for ESP8266, ESP8285 and ESP32项目地址https://gitcode.com/gh_mirrors/no/nodemcu-firmware点击查看免费下载相关推荐NodeMCU 固件 file 模块深度指南Lua 文件系统 API 全解析NodeMCU 固件 file 模块深度指南Lua 文件系统 API 全解析 NodeMCU 固件内置的 file 模块为 Lua 脚本提供了访问板载 SPI物联网嵌入式MQTT Explorer终极MQTT客户端完全指南MQTT Explorer终极MQTT客户端完全指南 MQTT Explorer是一款功能全面的MQTT客户端工具专为物联网开发者和系统管理员设计提供结构开发工具物联网消息队列上一篇终极指南如何用AI在3秒内完成Blender室内设计下一篇3分钟快速上手Boss Show Time - 你的智能招聘时间助手终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
NodeMCU 固件 MQTT 模块实战指南：基于 MQTT 3.1.1 的 Lua 客户端 API 全面解析
物联网嵌入式【免费下载链接】nodemcu-firmwareLua based interactive firmware for ESP8266, ESP8285 and ESP32项目地址https://gitcode.com/gh_mirrors/no/nodemcu-firmware点击查看免费下载本指南以 NodeMCU 固件内置的mqtt模块为主线系统讲解在 ESP8266/ESP8285/ESP32 平台上如何用 Lua 语言创建 MQTT 客户端、建立安全连接、订阅发布消息、处理遗嘱LWT与大消息溢出等完整场景并结合 app/modules/mqtt.c 源码剖析消息缓冲、心跳保活与 QoS 应答的实现原理。读完本文你将能够编写一套具备自动重连、遗嘱上报、按主题分发能力的完整 MQTT 应用。模块概述与协议版本要求mqtt模块是 NodeMCU 固件内置的 MQTT 客户端实现源码位于 app/modules/mqtt.c底层报文编解码依赖 app/mqtt 目录下的mqtt_msg.c、msg_queue.c组件。重要前提该客户端遵循MQTT 协议 3.1.1 版本。在使用前请确认你的 Broker如 Mosquitto、EMQX、CloudMQTT 等支持并正确配置为 3.1.1 版本该客户端与运行 MQTT 3.1 旧协议的 Broker 不兼容后向不兼容否则会在连接阶段直接收到CONNACK_REFUSED_PROTOCOL_VER错误。模块 API 概览API作用mqtt.Client()创建 MQTT 客户端对象mqtt:connect()连接 Brokermqtt:close()主动断开连接发送 DISCONNECTmqtt:lwt()配置 Last Will and Testament遗嘱消息mqtt:on()注册事件回调mqtt:publish()发布消息mqtt:subscribe()/mqtt:unsubscribe()订阅 / 取消订阅创建客户端mqtt.Client()语法与参数mqtt.Client(clientid, keepalive[, username, password, cleansession, max_message_length])参数说明clientid客户端 ID字符串。可省略源码中若省略会自动生成为NodeMCU_ 芯片 ID 十六进制串见 app/modules/mqtt.ckeepalive心跳保活间隔秒。源码中若传 0 或省略默认回退到 60 秒MQTT_DEFAULT_KEEPALIVE见 app/modules/mqtt.cusername用户名可选password密码可选cleansession0/1对应false/true。默认 1true即每次连接都建立全新会话max_message_length可接收的最大消息长度默认 1024 字节返回一个 MQTT 客户端对象。max_message_length 与内存安全根据 MQTT 规范PUBLISH 报文的理论上限是 256MB这对 NodeMCU 的 RAM 来说显然不现实。为了避免内存耗尽OOM固件强制限制可接收消息的大小该限制即由max_message_length参数控制默认值 1024 字节是刻意选定的——这是 NodeMCU 2.2.1 及更早版本中的隐含限制当时完全没有显式处理。“消息长度”指完整的 MQTT 报文大小包含固定头、可变头、主题名、包 ID如适用和 Payload。精确定义请参阅 MQTT v3.1.1 规范。实践中该参数只影响接收方向的 PUBLISH报文因为所有常规控制报文都很小。任何大于max_message_length的消息其能收到的部分会被投递到overflow回调如果已注册其余部分被丢弃后续消息不受影响。即使消息被丢弃如果 QoS 为 1 或 2固件仍会向 Broker 发送 ACK——对应源码中mqtt_msg_puback/mqtt_msg_pubrec的入队逻辑见 app/modules/mqtt.c。跨 TCP 包的大消息缓冲机制一个容易被忽视的内存细节即使单条消息没有超过max_message_length也可能触发堆内存分配。原因如下当一条消息跨越多个 TCP 包时固件会用堆内存缓冲不完整部分。从源码看固件在首次看到消息头时就按完整消息长度一次性分配calloc避免反复 realloc 造成堆碎片见 app/modules/mqtt.c。如果分配失败MQTT 会话会被直接断开。Broker 可能在同一个 TCP 包中连续发送多条小消息若最后一条消息在包内装不下固件也会分配堆缓冲等待下一个 TCP 包。单条消息能塞进一个 TCP 包的典型上限是1460 字节与固件内部MQTT_BUF_SIZE一致见 app/modules/mqtt.c但实际取决于网络 MTU 配置、分包情况以及同一包内是否混有多条消息。接收状态机在源码中体现为tReceiveState的四种状态MQTT_RECV_NORMAL、MQTT_RECV_BUFFERING_SHORT、MQTT_RECV_BUFFERING、MQTT_RECV_SKIPPING见 app/modules/mqtt.c。其中SKIPPING状态专门用于“超过上限的消息”固件会丢弃后续 TCP 包中的剩余字节而不做存储。基础示例-- 无登录信息keepalive 120 秒 m mqtt.Client(clientid, 120) -- 带登录信息keepalive 120 秒 m mqtt.Client(clientid, 120, user, password) -- 配置遗嘱消息可选 -- 若客户端在 keepalive 时间内未发送心跳包Broker 会向 /lwt 发布一条 -- qos 0、retain 0、payload offline 的消息 m:lwt(/lwt, offline, 0, 0) -- 离线事件 m:on(offline, function(client) print(offline) end) -- 收到消息事件 m:on(message, function(client, topic, data) print(topic .. :) if data ~ nil then print(data) end end) -- 溢出事件收到被截断的超限消息 m:on(overflow, function(client, topic, data) print(topic .. partial overflowed message: .. data) end) -- TLS 连接写法m:connect(192.168.11.118, secure-port, 1) m:connect(192.168.11.118, 1883, false, function(client) print(connected) -- 订阅与发布必须在连接建立成功后才进行 -- 可以放在这里connect 回调或通过跟踪连接状态保证时机。 -- 订阅主题QoS 0 client:subscribe(/topic, 0, function(client) print(subscribe success) end) -- 发布消息data hello, QoS 0, retain 0 client:publish(/topic, hello, 0, 0, function(client) print(sent) end) end, function(client, reason) print(Connection failed reason: .. reason) end) m:close() -- offline 回调触发后可以再次调用 m:connect注册事件回调mqtt:on()mqtt:on(event, function(client[, topic[, message]]))回调函数的第一个参数永远是客户端对象本身其余参数因事件而异。事件回调参数说明connectclient连接建立成功connfailclient, reason连接失败reason为失败码见下表subackclient订阅成功应答unsubackclient取消订阅成功应答pubackclient发布应答QoS 1/2或发送完成QoS 0messageclient, topic, message收到 PUBLISH 消息topic/message均为 Lua 字符串overflowclient, topic, message收到超限消息message被截断到max_message_lengthofflineclient已建立连接被关闭注意与connfail不同见下文源码中的事件名白名单与回调注册逻辑可在 app/modules/mqtt.c 查看connfail对应的回调槽位是cb_connect_fail_ref。一个需要留意的事实suback/unsuback/puback等事件不携带额外参数这带来两个后果——Broker 返回的“订阅允许的最大 QoS”信息会丢失如果你期望逐事件确认就必须自行管理消息队列源码中的pending_msg_q只是内部控制包队列并不暴露给 Lua 层。连接 Brokermqtt:connect()语法与参数mqtt:connect(host[, port[, secure]][, function(client)[, function(client, reason)]])参数说明hostBroker 主机名、域名或 IP字符串portBroker 端口数字默认 1883见 app/modules/mqtt.csecure布尔值true表示启用 TLSmqtts。请留意 net 模块 中记录的约束function(client)连接建立成功回调function(client, reason)连接失败回调失败后不应再调用其他回调返回nil结果一律通过回调观察。attention安全连接mqtts带有不少限制请务必阅读 tls 模块 文档中的警告。回调别名规则关键语义:connect()的第一个回调与:on(connect, ...)互为别名二者中最后传入者生效但若向:connect()传入nil已有的回调会被保留而非清除。第二个失败回调与:on(connfail, ...)互为别名。offline回调只在“已建立”的连接变为关闭时触发。如果connect()本身就没建立成功只会调用:connect()传入的失败回调offline不会被触发。这一语义在源码连接失败路径中体现为直接调用mqtt_connack_fail见 app/modules/mqtt.c。旧文档曾建议用整数 0/1传secure现在这样写会触发弃用警告请改用布尔值false/true源码中通过platform_print_deprecation_note提示见 app/modules/mqtt.c。连接失败回调原因码固件通过mqtt:on(connfail, ...)的第二个参数返回失败原因码这些常量定义在 app/mqtt/mqtt_msg.h并在模块加载时以mqtt.CONN_*常量导出见 app/modules/mqtt.c常量值描述mqtt.CONN_FAIL_SERVER_NOT_FOUND-5指定 IP 与端口上没有 Broker 在监听mqtt.CONN_FAIL_NOT_A_CONNACK_MSG-4Broker 的响应不是协议要求的 CONNACK 报文mqtt.CONN_FAIL_DNS-3DNS 解析失败mqtt.CONN_FAIL_TIMEOUT_RECEIVING-2等待 Broker CONNACK 超时mqtt.CONN_FAIL_TIMEOUT_SENDING-1发送 CONNECT 报文超时mqtt.CONNACK_ACCEPTED0无错误。注意该值不会触发失败回调mqtt.CONNACK_REFUSED_PROTOCOL_VER1Broker 不是 MQTT 3.1.1 Brokermqtt.CONNACK_REFUSED_ID_REJECTED2Broker 拒绝了指定的 ClientID参见mqtt.Client()mqtt.CONNACK_REFUSED_SERVER_UNAVAILABLE3服务器不可用mqtt.CONNACK_REFUSED_BAD_USER_OR_PASS4Broker 拒绝了指定的用户名或密码mqtt.CONNACK_REFUSED_NOT_AUTHORIZED5用户名未授权这些原因码与源码中的处理路径一一对应例如等待 CONNACK 超时对应MQTT_CONN_FAIL_TIMEOUT_RECEIVING见 app/modules/mqtt.c收到的不是 CONNACK 报文对应MQTT_CONN_FAIL_NOT_A_CONNACK_MSG见 app/modules/mqtt.cCONNACK 返回码非 0 则直接透传对应拒绝码见 app/modules/mqtt.c。实现可靠的自动重连应用应当监听连接失败并在错误回调中处理才能实现可靠的服务器连接。官方推荐的重试模式function handle_mqtt_error(client, reason) tmr.create():alarm(10 * 1000, tmr.ALARM_SINGLE, do_mqtt_connect) end function do_mqtt_connect() mqtt:connect(server, function(client) print(connected) end, handle_mqtt_error) end当然真正的项目中“connected”回调里应该做点有用的事订阅、发布、状态上报等。连接过程的底层调用链从源码看mqtt:connect()的完整链路是解析域名dns_gethostbyname失败返回MQTT_CONN_FAIL_DNS见 app/modules/mqtt.c→ 通过espconn_connect或 TLS 下的espconn_secure_connect建立 TCP 连接见 app/modules/mqtt.c→ TCP 连接建立后组装 CONNECT 报文并立即发送mqtt_msg_connect见 app/modules/mqtt.c→ 进入MQTT_CONNECT_SENDING/MQTT_CONNECT_SENT状态等待 CONNACK。整个过程有MQTT_SEND_TIMEOUT5 秒超时保护。配置遗嘱消息mqtt:lwt()Last Will and Testament遗嘱消息是 MQTT 的重要机制客户端异常掉线时由 Broker 代为发布一条预设消息常用于设备离线通知。mqtt:lwt(topic, message[, qos[, retain]])参数说明topic遗嘱发布主题字符串message遗嘱消息内容buffer 或字符串qosQoS 等级默认 0retainretain 标志默认 0遗嘱在连接时发送给 Broker因此lwt()必须在connect()之前调用。源码中遗嘱信息保存在conf结构的will_topic_ref/will_message_ref/will_qos/will_retain字段中并在建立 TCP 连接后随 CONNECT 报文一起组装见 app/modules/mqtt.c。Broker 何时发布遗嘱当 Broker 发现与客户端的连接中断时就会发布遗嘱消息。触发条件包括客户端在mqtt.Client()指定的 keepalive 时间内没有发送心跳包TCP 连接被正常关闭但在关闭前没有先关闭 MQTT 连接Broker 尝试向客户端发送数据时 TCP 连接断裂。例如你指定 keepalive 为 120 秒那么直接关掉设备电源、且 Broker 期间没有向该客户端推送任何数据时遗嘱消息会在掉线约120 秒后被发布。已知限制注意当前 NodeMCU MQTT 库存在一个 bug所有断连都会被表现为“意外断连”——MQTT 层的断开消息未在 TCP 连接拆除前发出。结果就是LWT 遗嘱消息几乎总是会被发布。跟踪见 nodemcu-firmware issue #3031。发布消息mqtt:publish()mqtt:publish(topic, payload, qos, retain[, function(client)])参数说明topic发布主题字符串遵循 MQTT 主题规范如a/b/c、通配符主题用于订阅而非发布payload消息内容buffer 或字符串qosQoS 等级0/1/2retainretain 标志0/1function(client)可选回调QoS 1/2 时收到 PUBACK 后触发QoS 0 时消息发送后触发返回true表示成功false表示失败。注意事项多次调用publish()时最后一次定义的回调会被所有 publish 命令调用。该回调参数与:on(puback, ...)互为别名。QoS 1 的应答在源码中通过MQTT_MSG_TYPE_PUBACK状态处理QoS 2 走完整的 PUBREC → PUBREL → PUBCOMP 四步握手见 app/modules/mqtt.c。源码确认QoS 0 消息在 TCP 发送完成mqtt_socket_sent时即触发回调并出队因为不会收到服务端 PUBACK见 app/modules/mqtt.c。订阅与取消订阅mqtt:subscribe()mqtt:subscribe(topic, qos[, function(client)]) mqtt:subscribe(table[, function(client)])参数说明topic主题字符串qos订阅 QoS 等级默认 0tabletopic, qos键值对数组一次订阅多个主题function(client)可选回调订阅成功后触发-- 订阅单主题QoS 0 m:subscribe(/topic, 0, function(conn) print(subscribe success) end) -- 一次订阅多个主题topic/0 → qos0; topic/1 → qos1; topic2 → qos2 m:subscribe({[topic/0]0, [topic/1]1, topic22}, function(conn) print(subscribe success) end)caution如果需要订阅多个主题不要多次调用subscribe()请使用上例所示的多主题 table 语法一次性订阅。多次调用subscribe()时最后一次定义的回调会被所有 subscribe 命令调用并与:on(suback, ...)互为别名。mqtt:unsubscribe()mqtt:unsubscribe(topic[, function(client)]) mqtt:unsubscribe(table[, function(client)])参数说明topic主题字符串tabletopic, anything键值对数组一次取消多个订阅值可为任意值源码只读取键function(client)可选回调取消订阅成功后触发-- 取消订阅单主题 m:unsubscribe(/topic, function(conn) print(unsubscribe success) end) -- 一次取消多个主题值可为任意值如 0 或 anything m:unsubscribe({[topic/0]0, [topic/1]0, topic2anything}, function(conn) print(unsubscribe success) end)同样地多次调用时最后一次定义的回调会被所有 unsubscribe 命令调用并与:on(unsuback, ...)互为别名。源码实现细节单主题/多主题订阅都经由mqtt_msg_subscribe_*系列函数在栈上临时缓冲区组装成 SUBSCRIBE 报文再整体入队发送若主题过多导致缓冲区溢出会返回buffer overflow, cant enqueue all subscriptions错误见 app/modules/mqtt.c。主动断开mqtt:close()mqtt:close()参数无返回nilclose()调度一次干净的连接拆除源码中它会先构造 DISCONNECT 报文入队并发送见 app/modules/mqtt.c。MQTT 协议要求客户端主动向服务器发送断开意愿以避免触发遗嘱消息。因此调用close()后客户端不能立即复用必须等到offline回调触发后才能再次connect()。综合实战云端 MQTT 收发示例以下示例来自仓库 lua_examples/mqtt/mqtt2cloud.lua演示了连接云 Broker、按主题分发消息、周期性上报的完整模式-- test with cloudmqtt.com local m_dis {} -- 按主题分发收到的消息 local function dispatch(m, t, pl) if pl ~ nil and m_dis[t] then m_dist end end local function topic1func(_, pl) print(get1: .. pl) end local function topic2func(_, pl) print(get2: .. pl) end do m_dis[/topic1] topic1func m_dis[/topic2] topic2func -- 创建客户端clientid, keepalive60, user, pass local m mqtt.Client(nodemcu1, 60, test, test123) m:on(connect, function(client) print(connection .. node.heap()) client:subscribe(/topic1, 0, function() print(sub done) end) client:subscribe(/topic2, 0, function() print(sub done) end) client:publish(/topic1, hello, 0, 0) client:publish(/topic2, world, 0, 0) end) m:on(offline, function() print(disconnect to broker...) print(node.heap()) end) m:on(message, dispatch) -- 连接host, port, secure m:connect(m11.cloudmqtt.com, 11214, 0) -- 每 10 秒上报一次时间 tmr.create():alarm(10000, tmr.ALARM_AUTO, function() local pl time: .. tmr.time() m:publish(/topic1, pl, 0, 0) end) end底层实现要点回顾保活机制keepalive计时由软件定时器mqtt_socket_timer驱动见 app/modules/mqtt.c。进入MQTT_DATA状态后若定时器到期且发送队列为空则发送 PINGREQ 心跳若上一次 PINGREQ 已发出但未收到 PINGRESPkeepalive_sent仍为 1则判定心跳失联并直接断开连接。发送队列所有控制报文先经msg_enqueue进入pending_msg_q队列再由mqtt_send_if_possible逐条发送见 app/modules/mqtt.c保证与 Broker 的应答严格按序交互。接收处理mqtt_socket_received解析可变长消息头按tReceiveState状态机处理正常/缓冲/跳过三种情形并在一个 TCP 包内循环处理多条消息READPACKET循环见 app/modules/mqtt.c。模块导出mqtt.Client及全部CONN_*/CONNACK_*常量通过 LROT 表导出到 Lua 命名空间见 app/modules/mqtt.c这也是 Lua 层能直接引用mqtt.CONN_FAIL_DNS等常量的原因。小结NodeMCU 的mqtt模块为资源受限的 ESP 系列芯片提供了完整的 MQTT 3.1.1 客户端能力从连接、订阅发布、遗嘱配置到超限消息的溢出保护均有清晰的 Lua API 与源码级保障。开发时请特别关注三点Broker 必须是 MQTT 3.1.1、遗嘱必须在 connect 之前配置、close()后须等待offline回调才能重连。结合max_message_length的合理配置默认 1024可根据业务 Payload 大小上调但需评估堆内存占用即可构建稳定可靠的设备端消息通道。赞分享物联网嵌入式【免费下载链接】nodemcu-firmwareLua based interactive firmware for ESP8266, ESP8285 and ESP32项目地址https://gitcode.com/gh_mirrors/no/nodemcu-firmware点击查看免费下载相关推荐NodeMCU 固件 file 模块深度指南Lua 文件系统 API 全解析NodeMCU 固件 file 模块深度指南Lua 文件系统 API 全解析 NodeMCU 固件内置的 file 模块为 Lua 脚本提供了访问板载 SPI物联网嵌入式MQTT Explorer终极MQTT客户端完全指南MQTT Explorer终极MQTT客户端完全指南 MQTT Explorer是一款功能全面的MQTT客户端工具专为物联网开发者和系统管理员设计提供结构开发工具物联网消息队列上一篇终极指南如何用AI在3秒内完成Blender室内设计下一篇3分钟快速上手Boss Show Time - 你的智能招聘时间助手终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
