MQTT 知识小册子

# 推荐格式：项目前缀 / 设备或分类 / 数据项
myhome/livingroom/temp        # 客厅温度
myhome/livingroom/humidity    # 客厅湿度
farm/greenhouse-01/soil       # 大棚 1 号土壤湿度
device/esp32-a1b2/status      # 设备在线状态

# QoS 1：两步确认
发布者 PUBLISH  ────────▶  Broker
发布者 ◀──────── PUBACK   # 收到确认，发布者才丢弃消息；没收到就重发

# QoS 2：四步握手，保证不重
发布者 PUBLISH  ────────▶  Broker
发布者 ◀──────── PUBREC   # Broker 已收下并记录
发布者 PUBREL   ────────▶  Broker   # 可以投递给订阅者了
发布者 ◀──────── PUBCOMP  # 全流程完成

# 1. 连接时就立好遗嘱：异常断开时代发 offline
连接参数: Last Will Topic   = myproj/esp32-001/status
          Last Will Payload = offline，也勾选 Retain

# 2. 连接成功后立刻报在线（保留消息）
PUBLISH myproj/esp32-001/status = online  (Retain)

# 3. 任何人订阅这个主题，立刻知道设备在不在线
SUBSCRIBE myproj/+/status

# ── 第一条命令：订阅（持续接收消息，按 Ctrl+C 停止）──
# -h 服务器地址  -p 端口  -u 用户名  -P 密码
# -i Client ID（自己随便起名，但要全网唯一，别和别人撞）
# -t 要订阅的主题（# 是通配符，收下 mytest 下所有消息）
# -v 打印时把主题也一起显示出来
mosquitto_sub -h mqttyyc.top -p 1883 \
  -u your_username -P your_password \
  -i myproj-cli-sub -t 'mytest/#' -v

# ── 第二条命令：发布一条消息（另开一个终端窗口执行）──
# -m 后面是消息内容；发完命令自动退出
mosquitto_pub -h mqttyyc.top -p 1883 \
  -u your_username -P your_password \
  -i myproj-cli-pub -t mytest/hello -m '你好 MQTT'

# 想要加密连接：把端口换成 8883，再加上系统 CA 证书路径：
#   -p 8883 --capath /etc/ssl/certs

Name:     mqttyyc
Host:     mqttyyc.top
Port:     8883   （勾选 SSL/TLS）
Username: your_username
Password: your_password

import paho.mqtt.client as mqtt  # 导入 MQTT 客户端库（先 pip install paho-mqtt 安装）

# 连接成功时自动调用的回调函数；rc 是返回码，0 表示成功（对照第 6 章）
def on_connect(client, userdata, flags, rc):
    print('已连接, rc =', rc)
    client.subscribe('mytest/#')  # 连上后立刻订阅：# 收下 mytest 下所有主题

# 每收到一条消息就自动调用的回调函数
def on_message(client, userdata, msg):
    # msg.topic 是主题；payload 是字节串，decode() 把它转成文字
    print(msg.topic, '->', msg.payload.decode())

client = mqtt.Client(client_id='myproj-py-001')  # 创建客户端，ID 每台设备唯一
client.username_pw_set('your_username', 'your_password')  # 换成账户中心注册的账号密码
client.tls_set()  # 开启 TLS 加密，配合下面的 8883 端口
client.on_connect = on_connect  # 登记两个回调函数
client.on_message = on_message
client.connect('mqttyyc.top', 8883)  # 连接 Broker（8883 是加密端口）
client.publish('mytest/hello', '你好 MQTT')  # 发布第一条消息
client.loop_forever()  # 常驻运行：维持连接并持续接收消息

import mqtt from 'mqtt';  // 引入 MQTT.js 库（先 npm install mqtt 安装）

// 浏览器 / HTTPS 页面必须用 wss:// 加密地址 8084，结尾 /mqtt 不能丢
const client = mqtt.connect('wss://mqttyyc.top:8084/mqtt', {
  username: 'your_username',  // 换成账户中心注册的账号
  password: 'your_password',  // 换成对应密码
});

client.on('connect', () => {      // 连接成功的回调
  client.subscribe('mytest/#');   // 订阅主题：# 收下 mytest 下所有消息
  client.publish('mytest/hello', '你好 MQTT');  // 发布一条试试
});
client.on('message', (topic, payload) => {  // 收到消息的回调
  console.log(topic, '->', payload.toString());  // 打印主题和内容
});

#include <WiFi.h>          // ESP32 官方 WiFi 库
#include <PubSubClient.h>  // MQTT 客户端库（Arduino IDE 库管理器搜 PubSubClient 安装）

WiFiClient wifiClient;     // 底层 TCP 连接对象
PubSubClient mqtt(wifiClient);  // MQTT 客户端，建立在上面的 TCP 之上

void setup() {  // 开机只执行一次
  WiFi.begin("你的WiFi", "WiFi密码");  // 连接家里 WiFi，换成自己的
  while (WiFi.status() != WL_CONNECTED) delay(300);  // 没连上就等，直到连上为止

  mqtt.setServer("mqttyyc.top", 1883);  // 指定 Broker 地址和端口
  while (!mqtt.connected()) {  // 连不上就一直重试
    // 三个参数：Client ID（每台设备必须唯一）、用户名、密码
    mqtt.connect("myproj-esp32-001", "your_username", "your_password");
    delay(500);
  }
  mqtt.subscribe("mytest/#");  // 订阅主题，准备收消息
  mqtt.publish("mytest/hello", "你好 MQTT");  // 发布第一条消息
}

void loop() {  // 之后反复循环执行
  mqtt.loop();  // 必须经常调用：收消息、发心跳全靠它
}

{
  "dev": "esp32-001",    // 设备标识，方便定位
  "type": "env",        // 数据类型，方便消费端分发
  "temp": 26.5,          // 数值直接用数字，别转字符串
  "humi": 61,
  "batt": 3.92,
  "ts": 1755600000      // 时间戳，去重和排序都靠它
}

# 设备：每 30 秒发布一次
PUBLISH farm/greenhouse-01/temp  {"v": 26.5, "ts": 1755600000}  QoS 0

# 服务端：一个通配订阅收下整个大棚
SUBSCRIBE farm/greenhouse-01/#

# 应用下发（QoS 1，怕丢）
PUBLISH myproj/esp32-001/cmd/set  {"led": true, "id": 42}

# 设备订阅自己的指令主题
SUBSCRIBE myproj/esp32-001/cmd/#

# 设备执行后回执（带上指令 id 方便对应）
PUBLISH myproj/esp32-001/cmd/set/ack  {"id": 42, "ok": true}

# 应用订阅回执
SUBSCRIBE myproj/esp32-001/cmd/+/ack

PLC / 机器人控制器
      │  现场总线或厂商接口（Modbus、PROFINET、MC、FOCAS…）
      ▼
IoT 网关 / 边缘上位机        ← 协议转换、缓存补发、主题映射都在这层
      │  MQTT（推荐 TLS 8883）
      ▼
Broker（mqttyyc.top）
      │
      ▼
上位监控 / MES / 看板、报警推送

factory/line1/plc-01/data       # PLC 数据上行：QoS 0，周期发布
factory/line1/plc-01/cmd        # MES 下发指令：QoS 1，怕丢
factory/line1/plc-01/cmd/ack    # PLC 执行回执：带上指令 id
factory/line1/robot-01/status   # 机器人状态：Retained + 遗嘱（见第 5 章）
factory/line1/robot-01/alarm    # 报警事件：QoS 1，触发即发

# device_template.py —— 改凭据即可运行（基于 paho-mqtt 2.x，1.x 见注释）
import json, os, time                # Python 自带库：JSON 编解码 / 环境变量 / 时间
import paho.mqtt.client as mqtt     # MQTT 客户端库，先 pip install paho-mqtt

# ========= 改成你自己的 =========
USERNAME  = os.getenv("MQTT_USER", "your_username")  # 优先读环境变量，否则用后面的默认值
PASSWORD  = os.getenv("MQTT_PASS", "your_password")  # 把默认值换成自己的账号密码
CLIENT_ID = "myproj-py-001"        # 在 Broker 上的身份证，每台设备必须唯一
PREFIX    = "myproj/py-001"        # 自己的主题前缀，避免和别人撞车
# ======================================

# 连接成功（或失败）时自动调用的回调
def on_connect(client, userdata, flags, reason_code, properties):
    # paho 1.x 用户：签名改为 (client, userdata, flags, rc)，创建行见下方注释
    if reason_code == 0:                # 0 表示连接成功（对照第 6 章返回码表）
        print("已连接")
        client.subscribe(PREFIX + "/cmd/#", qos=1)  # 订阅自己的指令主题，QoS 1 怕丢
        client.publish(PREFIX + "/status", "online", qos=1, retain=True)  # 报在线（保留消息，见第 5 章）
    else:
        print("连接失败:", reason_code)  # 失败时打印原因码好排查

# 每收到一条订阅的消息就自动调用
def on_message(client, userdata, msg):
    print("收到:", msg.topic, msg.payload.decode())  # payload 是字节，decode() 转文字
    if msg.topic.endswith("/cmd/set"):          # 收到的是“设置”指令 → 回执告诉对方已执行
        client.publish(PREFIX + "/cmd/set/ack",
                       json.dumps({"ok": True, "ts": int(time.time())}), qos=1)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)  # 创建客户端（2.x 写法）
# paho 1.x 用户：上一行改为 mqtt.Client(client_id=CLIENT_ID)
client.username_pw_set(USERNAME, PASSWORD)   # 设置登录账号密码
client.tls_set()                                # 开启加密；删掉这行则用明文 1883
client.will_set(PREFIX + "/status", "offline", qos=1, retain=True)  # 遗嘱：异常掉线时 Broker 代发 offline
client.on_connect = on_connect              # 登记回调函数
client.on_message = on_message
client.reconnect_delay_set(2, 30)               # 断线后 2 秒起重连，逐渐延长到 30 秒，防重连风暴
client.connect_async("mqttyyc.top", 8883, keepalive=45)  # 异步连接，每 45 秒发一次心跳
client.loop_start()                            # 后台线程维持连接、收发消息

try:
    n = 0
    while True:                             # 主循环：模拟传感器定时上报
        n += 1
        client.publish(PREFIX + "/data", json.dumps({
            "dev": CLIENT_ID, "temp": 25 + n % 5, "ts": int(time.time())}), qos=0)  # 上报用 QoS 0 轻快即可
        time.sleep(5)                           # 每 5 秒上报一次
except KeyboardInterrupt:                  # 按 Ctrl+C 退出时优雅收尾
    client.publish(PREFIX + "/status", "offline", qos=1, retain=True)  # 主动报离线，避免误触发遗嘱
    client.loop_stop()

# collector_template.py —— 服务端示例：把收到的消息存进 CSV 表格，Ctrl+C 退出
import csv, os, time                # Python 自带库：写 CSV / 环境变量 / 时间
import paho.mqtt.client as mqtt     # MQTT 客户端库

USERNAME  = os.getenv("MQTT_USER", "your_username")  # 换成自己的账号
PASSWORD  = os.getenv("MQTT_PASS", "your_password")  # 换成自己的密码
SUB_TOPIC = "myproj/#"                 # 订阅的主题：# 收下 myproj 前缀下所有消息，换成自己的前缀
CSV_FILE  = "messages.csv"           # 保存的文件名，用 Excel 就能打开

# 连接成功后自动订阅
def on_connect(client, userdata, flags, reason_code, properties):
    print("已连接，订阅:", SUB_TOPIC)
    client.subscribe(SUB_TOPIC, qos=1)  # QoS 1：断线重连后能补收期间消息

# 每收到一条消息就追加写入 CSV 一行
def on_message(client, userdata, msg):
    with open(CSV_FILE, "a", newline="", encoding="utf-8") as f:  # "a" = 追加模式，不覆盖旧数据
        csv.writer(f).writerow([time.strftime("%F %T"), msg.topic,  # 三列：时间、主题、内容
                                msg.payload.decode(errors="replace")])  # errors 防止脏字节导致报错
    print("已存:", msg.topic)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="myproj-collector-001")  # 创建客户端，ID 唯一
client.username_pw_set(USERNAME, PASSWORD)  # 登录凭据
client.tls_set()                          # 开启加密，配合 8883
client.on_connect = on_connect           # 登记回调
client.on_message = on_message
client.reconnect_delay_set(2, 30)          # 重连退避
client.connect("mqttyyc.top", 8883, keepalive=45)  # 连接 Broker
client.loop_forever()                    # 常驻运行，一直收消息存文件

// esp32_template.ino —— 改 WiFi 和凭据后直接烧录
#include <WiFi.h>          // WiFi 库
#include <PubSubClient.h>  // MQTT 库（Arduino IDE 库管理器安装）

// ========= 改成你自己的 =========
const char* WIFI_SSID = "你的WiFi";          // WiFi 名称
const char* WIFI_PASS = "WiFi密码";          // WiFi 密码
const char* MQTT_USER = "your_username";   // 账户中心注册的账号
const char* MQTT_PASS = "your_password";   // 对应密码
const char* CLIENT_ID = "myproj-esp32-001";   // 每台设备必须唯一
const char* PREFIX    = "myproj/esp32-001";  // 自己的主题前缀
// ================================

WiFiClient wifiClient;          // 底层 TCP 连接
PubSubClient mqtt(wifiClient);  // MQTT 客户端
unsigned long lastReport = 0;   // 记录上次上报时间，用于定时

// 连接家里 WiFi，没连上就一直等
void connectWifi() {
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  while (WiFi.status() != WL_CONNECTED) { delay(500); Serial.print("."); }
  Serial.println("\nWiFi 已连接");
}

// 收到订阅消息时的回调：topic 是主题，payload 是内容字节
void onMessage(char* topic, byte* payload, unsigned int len) {
  String t = String(topic), body = String((char*)payload).substring(0, len);  // 把字节转成字符串好处理
  if (t.endsWith("/cmd/set")) {                 // 收到开关灯指令：内容里有 true 就点亮
    bool on = body.indexOf("true") >= 0;
    digitalWrite(LED_BUILTIN, on ? HIGH : LOW);  // 控制开发板自带 LED
    mqtt.publish((String(PREFIX) + "/cmd/set/ack").c_str(),  // 回执：告诉应用端已执行
                 ("{\"ok\":true,\"led\":" + String(on ? "true" : "false") + "}").c_str());
  }
}

// 开机初始化，只跑一次
void setup() {
  Serial.begin(115200);            // 打开串口监视器方便看调试输出
  pinMode(LED_BUILTIN, OUTPUT);  // LED 引脚设为输出
  connectWifi();
  configTime(8 * 3600, 0, "ntp.aliyun.com");    // 联网校时（东八区），这样上报能带真实时间戳
  mqtt.setServer("mqttyyc.top", 1883);            // Broker 地址；要加密换 WiFiClientSecure + 8883
  mqtt.setCallback(onMessage);   // 登记收消息回调
  mqtt.setKeepAlive(45);           // 每 45 秒心跳一次
}

// 主循环，反复执行
void loop() {
  if (WiFi.status() != WL_CONNECTED) connectWifi();  // WiFi 断了就重连
  if (!mqtt.connected()) {          // MQTT 断了就重连
    // connect 参数较多：Client ID、账号、密码、遗嘱主题、遗嘱 QoS、遗嘱保留、遗嘱内容 offline
    // 遗嘱作用：设备异常掉线时 Broker 自动代发 offline（见第 5 章）
    if (mqtt.connect(CLIENT_ID, MQTT_USER, MQTT_PASS,
                     (String(PREFIX) + "/status").c_str(), 1, true, "offline")) {
      mqtt.publish((String(PREFIX) + "/status").c_str(), "online", true);  // 连上后报在线（保留消息）
      mqtt.subscribe((String(PREFIX) + "/cmd/#").c_str());  // 订阅自己的指令主题
      Serial.println("MQTT 已连接");
    }
  }
  mqtt.loop();  // 必须经常调用：收消息、发心跳全靠它
  if (millis() - lastReport > 5000) {             // 距上次上报超过 5 秒 → 上报一次
    lastReport = millis();
    // 手工拼 JSON：dev 设备名、rssi 信号强度、up 运行秒数、ts 时间戳
    String json = "{\"dev\":\"" + String(CLIENT_ID) + "\",\"rssi\":" + String(WiFi.RSSI()) +
                  ",\"up\":" + String(millis() / 1000) + ",\"ts\":" + String((long)time(nullptr)) + "}";
    mqtt.publish((String(PREFIX) + "/data").c_str(), json.c_str());
  }
}

<!-- 保存为 mqtt-debug.html，双击打开即用 -->
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>MQTT Web 调试</title>
<script src="https://unpkg.com/mqtt@5/dist/mqtt.min.js"></script>
<style>
  body { font: 14px/1.7 system-ui, sans-serif; max-width: 760px; margin: 24px auto; padding: 0 16px; }
  input, button { padding: 6px 10px; margin: 3px 2px; border: 1px solid #888; border-radius: 6px; }
  #log { height: 280px; overflow: auto; background: #12261b; color: #9fd8ae;
         padding: 10px; border-radius: 8px; white-space: pre-wrap; }
</style>
</head>
<body>
<h2>MQTT Web 调试页</h2>
<input id="user" placeholder="用户名">
<input id="pass" type="password" placeholder="密码">
<button onclick="doConnect()">连接</button>
<button onclick="client && client.end()">断开</button><br>
<input id="sub" value="mytest/#" size="18">
<button onclick="client.subscribe(sub.value)">订阅</button>
<input id="topic" value="mytest/hello" size="14">
<input id="msg" value="你好 MQTT" size="14">
<button onclick="client.publish(topic.value, msg.value)">发布</button>
<pre id="log"></pre>
<script>
let client;                              // MQTT 客户端对象，连接前是空的
const logEl = document.getElementById("log");       // 页面底部的日志区域
const log = (s) => { logEl.textContent += s + "\n"; };  // 往日志里追加一行

// 点“连接”按钮时执行
function doConnect() {
  client = mqtt.connect("wss://mqttyyc.top:8084/mqtt", {  // 浏览器必须用 wss 加密地址
    username: user.value,   // 从输入框读取账号密码
    password: pass.value,
    clientId: "web-debug-" + Math.random().toString(16).slice(2, 8),  // 随机 ID，避免和别人撞
    reconnectPeriod: 3000,  // 断线 3 秒后自动重连
  });
  client.on("connect", () => log("✔ 已连接"));           // 连接成功→日志提示
  client.on("message", (t, p) => log(t + " -> " + p.toString()));  // 收到消息→打印主题和内容
  client.on("error", (e) => log("✘ " + e));            // 出错→打印错误信息
}
</script>
</body>
</html>