mqtt — MQTT 客户端

模块名：mqtt

连接 MQTT Broker 做发布/订阅。只支持明文连接（通常 1883 端口），不支持 TLS。调用 start 后内部线程会自动重连，脚本不需要自己写重连逻辑。

除构造函数 creat 之外的函数在库表和对象元表里都注册了，所以 mqtt.publish(client, "t", "v") 和 client:publish("t", "v") 完全等价。下文签名统一写库函数形式，客户端对象作为第一个参数；用冒号写法时该参数由语法糖自动传入。

MQTT Topic 树与通配符、QoS 0/1/2 握手对比、Retain 保留消息机制

函数总览

库级函数（第一参数传 client）

函数	作用
`mqtt.creat(url, port, client_id, username, password)`	创建 MQTT 客户端对象
`mqtt.set_version(version)`	设置协议版本
`mqtt.set_keepalive(keepalive)`	设置 KeepAlive 秒数
`mqtt.set_cleansession(cleansession)`	设置 CleanSession 标志
`mqtt.reg(msg_type, callback)`	注册事件回调
`mqtt.start()`	启动连接
`mqtt.publish(topic, payload, [qos, retain])`	发布消息
`mqtt.subscribe(client, topicN, ...)`	批量订阅主题
`mqtt.unsubscribe(client, topicN, ...)`	批量取消订阅
`mqtt.is_linked()`	查询连接状态

对象方法（冒号写法）

上表除 mqtt.creat 外的每一项都有同名对象方法，例如 client:start()、client:publish(topic, payload)、client:subscribe("t", 0)，作用与对应的库函数一致。mqtt.creat 是构造函数，只能用库表形式调用。

常量

常量	可用于 `reg`	回调入参
`mqtt.CONNECTED`	是	无
`mqtt.DISCONNECTED`	是	无
`mqtt.RECVDATA`	是	`(topic, payload)`
`mqtt.SUBSCRIBED`	否，注册返回 `false`	—
`mqtt.UNSUBSCRIBED`	否，注册返回 `false`	—
`mqtt.PUBLISHED`	否，注册返回 `false`	—

后三个常量虽然存在，但底层没有对应的分发逻辑，注册了也不会被调用。

mqtt.creat

lua

mqtt.creat(url, port, client_id, username, password) --> client

创建 MQTT 客户端对象。5 个参数全部必填。

参数

#	参数	类型	必填	默认值	说明
1	`url`	string	是	—	Broker 地址（域名或 IP），不要带 `mqtt://` 前缀
2	`port`	integer	是	—	端口，通常 `1883`
3	`client_id`	string	是	—	客户端 ID，同一 Broker 上必须唯一
4	`username`	string	是	—	用户名，无认证传空串 `""`
5	`password`	string	是	—	密码，无认证传空串 `""`

返回值

类型	说明
userdata	客户端对象。必须用变量长期持有，被 GC 回收会断开连接

示例

lua

mqttc = mqtt.creat("broker.emqx.io", 1883, "lpplc_001", "", "")

说明

建议用全局变量或长生命周期的 local 持有，避免脚本执行完后对象被回收。

实用例程

lua

local deviceSn="<设备ID或凭据>"

mqttClient=mqtt.creat("192.168.1.100", 1885, deviceSn, "<名称>", "<名称>")--建立MQTT对象
mqtt.reg(mqttClient,mqtt.CONNECTED,function()--注册登陆成功回调函数
		print("连接成功")
	end
)
mqtt.reg(mqttClient,mqtt.DISCONNECTED,function()--注册网络断开事件
		print("网络断开")
	end
)
mqtt.reg(mqttClient,mqtt.RECVDATA,function(a,b)--注册接收到数据事件
		print("收到数据：",a,b)
	end
)
mqttClient:start()--启动MQTT协议栈

把上面的地址、主题、引脚替换成你项目里的即可。

mqtt.set_version

lua

mqtt.set_version(client, version)

设置 MQTT 协议版本，必须在 start 之前调用。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2	`version`	integer	是	`4`	`3` = MQTT 3.1，`4` = MQTT 3.1.1

返回值

无返回值。

说明

也可用冒号写法 client:set_version(4)。

mqtt.set_keepalive

lua

mqtt.set_keepalive(client, sec)
client:set_keepalive(sec)

设置 KeepAlive 心跳间隔，必须在 start 之前调用。这是唯一除 start 之外两种写法都能用的函数。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2	`sec`	integer	是	`60`	心跳秒数，按 u8 截断（最大 255）

返回值

无返回值。

实用例程

lua

--以下为水源地（水库）程序
xd=mqtt.creat("192.168.1.100", 1883, "<名称>", "<名称>", "<名称>")--建立MQTT对象
mqtt.reg(xd,mqtt.CONNECTED,function()--注册登陆成功回调函数
		print("xd连接成功")
		mqtt.subscribe(xd,"/sys/thing/property/post_reply/<设备ID>/dlinks_<设备ID>",0)--订阅消息
	end)
mqtt.reg(xd,mqtt.DISCONNECTED,function()--注册网络断开事件
		print("xd 网络断开")
	end
)
mqtt.reg(xd,mqtt.RECVDATA,function(a,b)--注册接收到数据事件
		print("收到xd数据：",a,b)		
	end
)
mqtt.set_keepalive(xd,30)
xd:start()--启动MQTT协议栈


dsfb1=timer.start(600000,function()--启动定时器600秒(10分钟)发布一个数据，测试时1分钟一次		
		local  time1=os.time()-28800
		--以下为数据过滤规则，达不要求的数据，不在上传到平台
		tb3={}	--清空表
		if  plc.df(0) > 0 then  tb3.syd_momentflow = {time=time1, value= plc.df(0) } end  --瞬时流量不限制
		if  plc.df(2) > 0 then   tb3.syd_totalflow = {time=time1, value=plc.dd(2)}   end --累计流量规则：大于0				
		if  plc.df(4) > 0   then  tb3.syd_waterlevel = {time=time1, value= plc.df(4)}  end --水库水位高度 m 
		--if  plc.df(6) > 0   then  tb3.syd_Reservoircapacity = {time=time1, value=plc.df(6) }    end --水库库容m3(暂时未要求)
		
		
		
		--后面可不断增加数据		
		print("<名称>",json.encode(tb3))			
		tb1={
			id=plc.dsd(37),version="1.0",sys= {ack= 1}, method="thing.property.post",
			params=tb3
		}
		
		print("json：",json.encode(tb1))		
		mqtt.publish(xd,"/sys/thing/property/post/<设备ID>/dlinks_<设备ID>",""..json.encode(tb1),0)
		
	end)

把上面的地址、主题、引脚替换成你项目里的即可。

mqtt.set_cleansession

lua

mqtt.set_cleansession(client, flag)

设置 CleanSession 标志，必须在 start 之前调用。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2	`flag`	integer	是	`1`	`1` 每次连接清空会话，`0` 保留会话

返回值

无返回值。

mqtt.reg

lua

mqtt.reg(client, event, callback) --> boolean

注册事件回调。要在 start 之前把 RECVDATA 注册好，否则可能漏掉早期消息。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2	`event`	integer	是	—	`mqtt.CONNECTED` / `mqtt.DISCONNECTED` / `mqtt.RECVDATA`
3	`callback`	function	是	—	回调函数。类型不是 function 会直接报错

返回值

类型	说明
boolean	是否注册成功。`SUBSCRIBED`/`UNSUBSCRIBED`/`PUBLISHED` 返回 `false`

示例

lua

mqtt.reg(mqttc, mqtt.CONNECTED, function()
    print("已连接")
    mqtt.subscribe(mqttc, "cmd/lpplc_001", 0)
end)

mqtt.reg(mqttc, mqtt.RECVDATA, function(topic, payload)
    print("收到", topic, payload)
end)

实用例程

lua

local deviceSn="<设备ID或凭据>"

mqttClient=mqtt.creat("192.168.1.100", 1885, deviceSn, "<名称>", "<名称>")--建立MQTT对象
mqtt.reg(mqttClient,mqtt.CONNECTED,function()--注册登陆成功回调函数
		print("连接成功")
	end
)
mqtt.reg(mqttClient,mqtt.DISCONNECTED,function()--注册网络断开事件
		print("网络断开")
	end
)
mqtt.reg(mqttClient,mqtt.RECVDATA,function(a,b)--注册接收到数据事件
		print("收到数据：",a,b)
	end
)
mqttClient:start()--启动MQTT协议栈
t=timer.start(3000,function()--定时发送 
		if mqtt.is_linked(mqttClient)   then
			local data =
			{
				device_id=deviceSn,
				device_status=plc.d(0),
				product_count = plc.dd(200)
			}
			pushData(data)
		end
	end
)
function push(t, s) if mqtt.is_linked(mqttClient) then mqtt.publish(mqttClient, t, s, 0) end end

把上面的地址、主题、引脚替换成你项目里的即可。

mqtt.start

lua

mqtt.start(client)
client:start()

启动连接。内部线程会持续尝试连接和重连，本函数立即返回，不等待连接建立。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象

返回值

无返回值。

说明

连接是否真的建好要靠 mqtt.CONNECTED 回调或 mqtt.is_linked 判断。

mqtt.publish

lua

mqtt.publish(client, topic, payload, [qos, retain]) --> boolean | nil

发布一条消息。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2	`topic`	string	是	—	主题
3	`payload`	string	是	—	消息内容，可为二进制
4	`qos`	integer	否	`0`	QoS 等级 `0`/`1`/`2`
5	`retain`	integer	否	`0`	`1` 让 Broker 保留该消息

返回值

类型	说明
boolean	发布是否成功
nil	未连接时直接返回 nil（不发送，也不返回 false）

示例

lua

mqtt.publish(mqttc, "data/lpplc_001", json.encode({ temp = plc.df(100) }), 0, 0)

说明

判断失败应写 if not mqtt.publish(...) then，不能写 == false，因为未连接时返回的是 nil。

实用例程

lua

mqttClient:start()--启动MQTT协议栈
t=timer.start(3000,function()--定时发送 
		if mqtt.is_linked(mqttClient)   then
			local data =
			{
				device_id=deviceSn,
				device_status=plc.d(0),
				product_count = plc.dd(200)
			}
			pushData(data)
		end
	end
)
function push(t, s) if mqtt.is_linked(mqttClient) then mqtt.publish(mqttClient, t, s, 0) end end

function pushData(data)
	local j = json.encode(data)
	push("device/machine".. deviceSn.."/status", j)
end

把上面的地址、主题、引脚替换成你项目里的即可。

lua

mqtt.subscribe(client, topic1, qos1, [topic2, qos2, ...]) --> boolean

一次订阅一个或多个主题，主题和 QoS 成对出现。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2, 4, 6…	`topic`	string	是	—	主题，支持 `+` 单层和 `#` 多层通配符
3, 5, 7…	`qos`	integer	是	—	对应主题的 QoS

返回值

类型	说明
boolean	订阅报文是否发送成功

示例

lua

mqtt.subscribe(mqttc, "cmd/lpplc_001", 0, "config/#", 1)

说明

本函数不检查连接状态，未连接时调用会失败。建议放在 mqtt.CONNECTED 回调里。

实用例程

lua

--以下为原水程序
xd=mqtt.creat("192.168.1.100", 1883, "<名称>", "<名称>", "<名称>")--建立MQTT对象
mqtt.reg(xd,mqtt.CONNECTED,function()--注册登陆成功回调函数
		print("xd连接成功")
		mqtt.subscribe(xd,"/sys/thing/property/post_reply/<设备ID>/dlinks_<设备ID>",0)--订阅消息
	end)
mqtt.reg(xd,mqtt.DISCONNECTED,function()--注册网络断开事件
		print("xd 网络断开")
	end
)
mqtt.reg(xd,mqtt.RECVDATA,function(a,b)--注册接收到数据事件
		print("收到xd数据：",a,b)		
	end
)
xd:start()--启动MQTT协议栈


dsfb1=timer.start(600000,function()--启动定时器600秒(10分钟)发布一个数据，测试时1分钟一次		
		local  time1=os.time()-28800
		--所有数据用string.format("%011.2f", V)格式化数据（总位数10位，小数2位，lua包含小数），以满足市局平台要求			
		tb1={
			id=plc.dsd(37),version="1.0",sys= {ack= 1}, method="thing.property.post",
			params={					
				ys_wt={time=time1, value= plc.df(6)},--温度
				ys_ph={time=time1, value= plc.df(4)},--PH值
				ys_turb={time=time1, value=plc.df(8) },--浊度
				--ys_pressure={time=time1, value=string.format("%011.2f", plc.d(14)/100) },	--原水压力
				ys_momentflow={time=time1, value=plc.df(0) },	--瞬时流量
				ys_totalflow={time=time1, value=plc.dd(2) },	--累计流量		
			}
		}
		
		print("json：",json.encode(tb1))		
		mqtt.publish(xd,"/sys/thing/property/post/<设备ID>/dlinks_<设备ID>",""..json.encode(tb1),0)
		
	end)

把上面的地址、主题、引脚替换成你项目里的即可。

mqtt.unsubscribe

lua

mqtt.unsubscribe(client, topic1, [topic2, ...]) --> boolean

取消订阅一个或多个主题。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象
2…	`topic`	string	是	—	要取消的主题

返回值

类型	说明
boolean	未连接时返回 `false`；否则返回报文是否发送成功

版本限制：2026-08-16 之前编译的固件调用会抛错

那批固件把主题参数的下标写错了一位，调用会报：

bad argument #1 to 'unsubscribe' (string expected, got userdata)

之后的版本已修正，正常返回结果。分不清手上设备的固件日期就用 pcall 包一次，报上面这个错就是旧固件。

变通做法（新旧固件都成立）：在 RECVDATA 回调里对不再关心的主题直接 return 忽略；或断开后用 set_cleansession(client, 1) 重连，让 Broker 清掉旧订阅。

mqtt.is_linked

lua

mqtt.is_linked(client) --> integer

查询当前连接状态。

参数

#	参数	类型	必填	默认值	说明
1	`client`	userdata	是	—	客户端对象

返回值

类型	说明
integer	`1` 已连接，`0` 未连接。注意是整数，不是 boolean

说明

因为返回整数，if mqtt.is_linked(c) then 在未连接时也成立（Lua 里 0 为真）。必须写 if mqtt.is_linked(c) == 1 then。

实用例程

lua

local deviceSn="<设备ID或凭据>"

mqttClient=mqtt.creat("192.168.1.100", 1885, deviceSn, "<名称>", "<名称>")--建立MQTT对象
mqtt.reg(mqttClient,mqtt.CONNECTED,function()--注册登陆成功回调函数
		print("连接成功")
	end
)
mqtt.reg(mqttClient,mqtt.DISCONNECTED,function()--注册网络断开事件
		print("网络断开")
	end
)
mqtt.reg(mqttClient,mqtt.RECVDATA,function(a,b)--注册接收到数据事件
		print("收到数据：",a,b)
	end
)
mqttClient:start()--启动MQTT协议栈
t=timer.start(3000,function()--定时发送 
		if mqtt.is_linked(mqttClient)   then
			local data =
			{
				device_id=deviceSn,
				device_status=plc.d(0),
				product_count = plc.dd(200)
			}
			pushData(data)
		end
	end
)
function push(t, s) if mqtt.is_linked(mqttClient) then mqtt.publish(mqttClient, t, s, 0) end end

function pushData(data)
	local j = json.encode(data)
	push("device/machine".. deviceSn.."/status", j)
end

把上面的地址、主题、引脚替换成你项目里的即可。

例程

每个例程都是可直接下载运行的完整脚本：注册完回调与定时器后脚本就结束，固件自动进入消息泵。 API 只用本页已文档化的函数；统一模板「场景 / 完整代码 / 要点 / 验证方法 / 注意事项」。 Broker 用公网 broker.emqx.io 也行，但同一 Broker 上 client_id 必须唯一 —— 两台设备用同一个 ID 会互相顶下线，抄脚本时记得改成序列号或设备名。

例程 1：定时上报本地数据，并把在线状态镜像到 M100

适用：LP3201 / LP3202 / LP3203 / LP3222 / LP3232 / LP3288 / 软PLC模拟器｜ 依赖模块：mqtt json timer plc ｜难度：★☆☆

场景：每 10 秒把温度、输出状态打包成 JSON 发到上行主题 lp3232/dev01/uplink；同时把 MQTT 的连接状态写进 M100，梯形图和触摸屏画面可以直接拿这个位做「云在线」指示灯，不用为了显示再单独写一段协调逻辑。

完整代码（可直接下载运行）：

lua

-- 定时上报 + 连接状态镜像到 M100（给梯形图 / 触摸屏用）
local PUB = "lp3232/dev01/uplink"

mqttc = mqtt.creat("192.168.1.20", 1883, plc.plc_name(), "", "")  -- 必须长期持有（全局）
mqtt.set_keepalive(mqttc, 60)                                     -- 必须在 start 之前
plc.m(100, 0)

mqtt.reg(mqttc, mqtt.CONNECTED, function()
    plc.m(100, 1)
    print("MQTT 已连接")
end)

mqtt.reg(mqttc, mqtt.DISCONNECTED, function()
    plc.m(100, 0)
    print("MQTT 已断开，内部线程会自动重连")
end)

mqtt.start(mqttc)                                                 -- 立即返回，连接后台建立

local seq = 0
timer.start(10000, function()
    seq = seq + 1
    local msg = json.encode({
        dev  = plc.plc_name(),
        seq  = seq,
        temp = plc.df(100),                    -- 浮点用 df()，plc.DF100 读出来是 nil
        y0   = plc.y(0),
        ts   = os.time()
    })

    if not mqtt.publish(mqttc, PUB, msg, 0, 0) then               -- 未连接时返回 nil，不是 false
        print("上报失败 #", seq)
    else
        print("已上报 #", seq, msg)
    end
end)

要点：

mqttc 必须是全局或长生命周期变量：被 GC 回收会直接断开连接，典型表现是「连上一会儿就没了」。
set_keepalive / set_version / set_cleansession 都要在 start 之前调用，之后再改无效。
start 之后不必自己写重连逻辑，内部线程会持续重试；用 CONNECTED / DISCONNECTED 回调感知状态即可。
判成功要写 if not mqtt.publish(...) then —— 未连接时返回的是 nil，写 == false 判断不到。
plc.m(100, 1/0) 就是给梯形图看的「云在线」位，约定见 Lua 与 PLC 协同开发。

验证方法：

先在 PC 上确认 Broker 可达（同网段起一个 mosquitto，或用公网 broker.emqx.io）。
下载脚本、重启设备，日志打印 MQTT 已连接；上位机「在线监视」里 M100 为 ON。
每 10 秒一条 已上报 #1 {"dev":...,"temp":25.3,...}；用 MQTTX 之类客户端订阅 lp3232/dev01/uplink 应收到同样内容。
拔网线：日志打印 MQTT 已断开、M100 转 OFF、上报失败；插回后自动重连、M100 再次 ON（这是 DISCONNECTED 回调有效的最直观证据）。

注意事项：

mqtt.is_linked() 返回的是整数：if mqtt.is_linked(c) then 在未连接时也为真（Lua 里 0 是真值），必须写 == 1，对照写法见例程 3。
client_id 重复会让两台设备互相顶下线，别在多台设备上原样抄同一份脚本。
plc.df(100) 读浮点、plc.d(100) 只读 16 位整数，两者别混。

其余例程（各自一页）

注意事项

库函数写法 mqtt.函数(client, ...) 与冒号写法 client:函数(...) 等价，creat 除外（构造函数只能用库表形式）。
is_linked 返回整数，判断连接必须 == 1。
set_version / set_keepalive / set_cleansession 都要在 start 之前调用，之后再改无效。
客户端对象必须长期持有；用局部变量在脚本执行完后会被 GC，连接随之断开。
只支持明文 MQTT，不支持 TLS。crypto 模块不能替代 TLS：它提供的是 AES 分组加密与 MD5，没有证书、握手与会话密钥协商，broker 侧也认不了（见 A-A 速查的 crypto 节，该模块本身还只在部分机型上提供）。设备侧能用的只有 plc.aes_encode / plc.aes_decode（密钥固化在设备里，且不足 16 字节不加密、非 16 整数倍的尾部原样明文），不能当传输层加密，只能算 payload 混淆。要真正的机密性请走云服务的加密通道或改走内网。

5.1 HTTP 客户端 (http)

5.2 MQTT 客户端 (mqtt)

5.3 TCP 客户端 (tcp_client)

5.4 TCP 服务器 (tcp_server)

5.5 网络工具 (net)

5.6 UDP 通信 (udp)

mqtt — MQTT 客户端

函数总览

库级函数（第一参数传 client）

对象方法（冒号写法）

常量

mqtt.creat

实用例程

mqtt.set_version

mqtt.set_keepalive

实用例程

mqtt.set_cleansession

mqtt.reg

实用例程

mqtt.start

mqtt.publish

实用例程

实用例程

mqtt.unsubscribe

mqtt.is_linked

实用例程

例程

例程 1：定时上报本地数据，并把在线状态镜像到 M100

其余例程（各自一页）

注意事项

mqtt — MQTT 客户端 ​

函数总览 ​

库级函数（第一参数传 client） ​

对象方法（冒号写法） ​

常量 ​

mqtt.creat ​

实用例程 ​

mqtt.set_version ​

mqtt.set_keepalive ​

实用例程 ​

mqtt.set_cleansession ​

mqtt.reg ​

实用例程 ​

mqtt.start ​

mqtt.publish ​

实用例程 ​

mqtt.subscribe ​

实用例程 ​

mqtt.unsubscribe ​

mqtt.is_linked ​

实用例程 ​

例程 ​

例程 1：定时上报本地数据，并把在线状态镜像到 M100 ​

其余例程（各自一页） ​

注意事项 ​

mqtt — MQTT 客户端

函数总览

库级函数（第一参数传 client）

对象方法（冒号写法）

常量

mqtt.creat

实用例程

mqtt.set_version

mqtt.set_keepalive

实用例程

mqtt.set_cleansession

mqtt.reg

实用例程

mqtt.start

mqtt.publish

实用例程

mqtt.subscribe

实用例程

mqtt.unsubscribe

mqtt.is_linked

实用例程

例程

例程 1：定时上报本地数据，并把在线状态镜像到 M100

其余例程（各自一页）

注意事项