2.5 对接 MQTT 收发 JSON

问题场景

既要把设备数据以 JSON 格式发到 MQTT 服务器（EMQX / 阿里云 IoT / 自建 Mosquitto），后端也要能通过 MQTT 下发指令让设备执行，上下行都需要打通。

做完之后能得到什么

• 上行：MQTTX 订阅数据 topic，每隔几秒收到设备发来的 JSON
• 下行：MQTTX 往指令 topic 发一条 JSON，设备立刻响应（Y0 灯亮 / 寄存器变化），并回一条 ack

需要准备

• 设备已联网
• 一个 MQTT 服务器（自建 Mosquitto / EMQX，或阿里云 IoT）
• MQTTX 客户端（用来验证）

预计耗时：25 分钟

动手做

第 1 步 · 配置 MQTT 连接

在上位机的 MQTT 配置页填写：

参数	说明
服务器地址	broker 域名或 IP
端口	1883（明文 TCP，固件不支持 TLS）
ClientID	全局唯一，建议用设备序列号
用户名 / 密码	按 broker 要求填
保活时间	60 秒

第 2 步 · 定义 Topic 规范

先把命名定下来，后面扩展才不会乱：

上行数据：linpan/{device_id}/data
下行指令：linpan/{device_id}/cmd
指令回执：linpan/{device_id}/ack

第 3 步 · 上行 · 连接并发布 JSON

mqtt 模块的函数既可写成 mqtt.函数(client, ...)，也可用冒号写法 client:函数(...)，两者等价（creat 除外）。客户端对象要用全局变量长期持有，否则会被 GC 回收导致断连。

-- 5 个参数全部必填；无认证时用户名/密码传空串
mqttc = mqtt.creat("broker.emqx.io", 1883, "LP3202-001", "", "")
mqtt.set_keepalive(mqttc, 60)
mqtt.start(mqttc)

-- 每 5 秒上报一次。未连接时 publish 返回 nil，先用 is_linked 判断（返回整数，要 == 1）
timer.start(5000, function()
  if mqtt.is_linked(mqttc) == 1 then
    local msg = {
      device_id = "LP3202-001",
      ts        = os.time(),
      temp      = plc.d(100) / 10,   -- D100 存放放大 10 倍的温度整数
      y0        = plc.y(0)           -- 读 Y0 输出状态
    }
    mqtt.publish(mqttc, "linpan/LP3202-001/data", json.encode(msg), 1, 0)
  end
end)

先只跑这段，确认 MQTTX 能收到数据，再往下加下行处理。

第 4 步 · 下行 · 解析 JSON 并执行

下行回调必须在 mqtt.start 之前用 mqtt.reg 注册；订阅要放在 CONNECTED 回调里（未连接时订阅会失败）。下面是上下行合在一起的完整脚本，直接整段复制即可运行：

-- 客户端对象用全局变量长期持有，避免被 GC 回收
mqttc = mqtt.creat("broker.emqx.io", 1883, "LP3202-001", "", "")
mqtt.set_keepalive(mqttc, 60)

-- 连接成功后再订阅下行指令主题
mqtt.reg(mqttc, mqtt.CONNECTED, function()
  mqtt.subscribe(mqttc, "linpan/LP3202-001/cmd", 1)
end)

-- 下行：解析指令并执行。回调在主循环里跑，别做长阻塞操作
mqtt.reg(mqttc, mqtt.RECVDATA, function(topic, payload)
  -- 必须用 pcall 包住，字段缺失时不会把脚本搞崩
  local ok, cmd = pcall(json.decode, payload)
  if not ok or type(cmd) ~= "table" or not cmd.action then
    return
  end

  if cmd.action == "set_y" then
    plc.y(cmd.no, cmd.value)       -- 写 Y 输出，value 传 0/1
  elseif cmd.action == "set_reg" then
    plc.d(cmd.addr, cmd.value)     -- 写 D 寄存器
  end

  -- 回执，让后端知道指令已执行
  mqtt.publish(mqttc, "linpan/LP3202-001/ack", json.encode({
    req_id = cmd.req_id,
    ok     = true,
    ts     = os.time()
  }), 1, 0)
end)

mqtt.start(mqttc)

-- 上行：每 5 秒上报一次
timer.start(5000, function()
  if mqtt.is_linked(mqttc) == 1 then
    mqtt.publish(mqttc, "linpan/LP3202-001/data", json.encode({
      device_id = "LP3202-001",
      ts        = os.time(),
      temp      = plc.d(100) / 10,
      y0        = plc.y(0)
    }), 1, 0)
  end
end)

第 5 步 · 用 MQTTX 联调

验证上行：MQTTX 订阅 linpan/LP3202-001/data，应看到持续到来的 JSON。

验证下行：MQTTX 发布到 linpan/LP3202-001/cmd：

{ "req_id": "abc123", "action": "set_y", "no": 0, "value": 1 }

验证成功

• 上行：MQTTX 里连续看到格式正确的 JSON
• 下行：发指令后 Y0 灯亮，并在 ack topic 收到回执

注意事项

WARNING

• 调用方式：mqtt.publish(mqttc, ...) 与 mqttc:publish(...) 两种写法等价（creat 除外），按习惯选一种即可
• 客户端对象要长期持有：用 local mqttc 时脚本执行完会被 GC 回收，连接随之断开，应改用全局变量
• is_linked 返回整数：必须写 == 1。直接写 if mqtt.is_linked(c) then 在未连接时（返回 0）也成立
• 未连接时 publish 返回 nil：判断失败要写 if not mqtt.publish(...) then，不能写 == false
• 订阅要放在 CONNECTED 回调里：subscribe 不检查连接状态，连上之前调用会失败
• RECVDATA 要在 start 之前注册：否则可能漏掉早期消息
• 只支持明文 MQTT：不支持 TLS，8883 端口无法连接；需要加密时用 crypto 自行处理 payload
• JSON 字段缺失会导致脚本中断：应用 pcall 包住 json.decode，并检查字段是否存在
• QoS 选择：推荐上行 QoS 1，指令 QoS 1 配合 req_id 去重
• Retain 谨慎开启：会让新客户端连上时立即收到一条旧数据，容易误动作
• 时间戳统一 UTC 秒：混用本地时间会导致跨时区数据错乱
• 指令必须幂等：用 req_id 记录已处理过的指令，防止网络重发导致重复动作
• 权限校验：不应允许任意客户端下发指令，建议在 broker 侧配置 Topic ACL
• ClientID 重复：两台设备用同一个 ClientID 会互相挤断对方的连接

举一反三

• 上报报警事件、设备心跳
• 远程设定 PID 参数、批量下发配方
• 一台设备汇总多路 Modbus 从站数据后统一上报
• 不想维护长连接？改用 HTTP → 2.6 对接 HTTP 上报与拉取数据

想更深入

• 服务器数据对接 · MQTT 对接
• Lua 编程