当前位置：首页 > 技术分享 > 正文内容

MQTT从入门到实战：物联网通信协议的全面解析与上手攻略

MQTT 从入门到实战

MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅消息传输协议，专为低带宽、高延迟、不可靠的网络环境设计。自1999年由IBM开发以来，已成为物联网领域的事实标准协议。本文将从核心概念讲起，带你一步步搭建属于自己的 MQTT 通信系统。

一、MQTT 核心概念

1. 发布/订阅模型

与传统的客户端/服务器请求-响应模式不同，MQTT 采用发布/订阅（Pub/Sub）模型。发送消息的一方（发布者）和接收消息的一方（订阅者）不需要直接连接，而是通过一个中间代理（Broker）进行消息转发。

💡 打个比方：就像关注微信公众号——你（订阅者）关注了某个号，号主（发布者）发文后，微信服务器（Broker）自动推送到你手上。发布者不需要知道谁在关注，订阅者也不需要知道号主是谁。

2. 三大核心角色

角色	说明	举例
Broker（代理）	消息中转服务器，接收并转发消息	EMQX、Mosquitto
Publisher（发布者）	发送消息到特定主题的设备/程序	温湿度传感器、手机App
Subscriber（订阅者）	订阅特定主题，接收消息的设备/程序	控制面板、数据库记录器

3. 主题（Topic）

主题是 MQTT 消息的路由标签，采用层级结构，用 / 分隔：

temp/room1 — 1号房间温度
sensor/greenhouse/humidity — 大棚湿度
cmd/led/1 — 控制1号LED开关

通配符订阅：

+ — 匹配单层：temp/+/humidity 可匹配 temp/room1/humidity 和 temp/room2/humidity
# — 匹配多层：sensor/# 可匹配 sensor/temp/room1、sensor/humidity 等所有以 sensor/ 开头的主题

二、QoS 质量等级

MQTT 定义了三种消息服务质量等级，按可靠性递增：

等级	名称	工作机制	适用场景
0	至多一次	发完即忘，不确认	传感器数据上报（丢一帧无所谓）
1	至少一次	确保送达，可能重复	控制指令（重复执行没关系）
2	恰好一次	四次握手，确保不重复	支付指令、关键告警（不能重复）

⚠️ 实战建议：大多数 IoT 场景用 QoS 1 就够了。QoS 2 开销大、速度慢，非必要不启用。QoS 0 适合高频传感器数据，比如每秒上报一次的温度值。

三、MQTT 的特别特性

保留消息（Retained Message）

发布者发送消息时可以标记为保留。Broker 会保存这条消息，新的订阅者上线后立即收到最后一条保留消息。这对状态同步非常有用——新设备接入后马上知道当前状态，不用等下一次上报。

遗嘱消息（Last Will）

设备连接时可以设置一份"遗嘱"。当设备非正常断开（断电、断网）时，Broker 会自动代发遗嘱消息，通知其他设备。这在设备监控中非常实用——可以立即发现哪些设备掉线了。

心跳保活（Keep Alive）

客户端定期发送 PINGREQ 报文通知 Broker 自己还在线。如果 Broker 在超时时间内没收到心跳，就认为客户端断连，触发遗嘱机制。推荐设置为 30~60 秒。

四、搭建 MQTT 服务器（以 EMQX 为例）

EMQX 是目前最流行的开源 MQTT Broker 之一，支持百万级并发连接。我们服务器上就运行着 EMQX 5.7.2。

Docker 一键部署

docker run -d --name emqx \
  -p 1883:1883 -p 8083:8083 -p 8084:8084 \
  -p 8883:8883 -p 18083:18083 \
  emqx/emqx:5.7.2

端口	协议	用途
1883	MQTT	明文 MQTT 连接（不加密）
8883	MQTTS	SSL/TLS 加密 MQTT 连接
8083	WS	WebSocket 连接（浏览器用）
8084	WSS	加密 WebSocket 连接
18083	HTTP	EMQX Dashboard 管理面板

部署完成后，访问 http://服务器IP:18083 即可进入 Dashboard（默认账号 admin，密码 public）。

五、MQTT 客户端工具推荐

1. MQTTX（推荐）

跨平台桌面客户端，支持 MQTT/MQTTS/WS/WSS，界面简洁，支持多主题订阅和消息历史。适合调试和测试。
官网：https://mqttx.app

2. Mosquitto_pub / mosquitto_sub

Mosquitto 项目自带的命令行工具，适合脚本集成：

# 订阅主题
mosquitto_sub -h so.lmwmm.com -t "sensor/#" -v

# 发布消息
mosquitto_pub -h so.lmwmm.com -t "cmd/led/1" -m "ON"

3. Python Paho-MQTT

Python 生态最流行的 MQTT 库，适合后端集成：

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print("已连接")
    client.subscribe("sensor/#")

def on_message(client, userdata, msg):
    print(f"{msg.topic}: {msg.payload.decode()}")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("so.lmwmm.com", 1883, 60)
client.loop_forever()

六、实战：ESP8266 连接 EMQX

以下是一个 ESP8266 通过 MQTT 上报温湿度、接收控制指令的完整示例。使用 PubSubClient 库（Arduino IDE 库管理器中搜索安装）。

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

// WiFi 配置
const char* ssid = "你的WiFi名称";
const char* password = "你的WiFi密码";

// MQTT 配置
const char* mqtt_server = "so.lmwmm.com";
const int mqtt_port = 1883;
const char* mqtt_user = "esp8266_01";
const char* mqtt_pass = "你的密码";

// 主题
const char* topic_temp = "sensor/esp01/temp";
const char* topic_hum  = "sensor/esp01/hum";
const char* topic_cmd  = "cmd/esp01/led";

WiFiClient espClient;
PubSubClient client(espClient);

void setup() {
  Serial.begin(115200);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) delay(500);
  client.setServer(mqtt_server, mqtt_port);
  client.setCallback(callback);
}

void callback(char* topic, byte* payload, unsigned int length) {
  String msg = "";
  for (int i = 0; i < length; i++) msg += (char)payload[i];
  if (String(topic) == topic_cmd) {
    if (msg == "ON") digitalWrite(LED_BUILTIN, LOW);
    else digitalWrite(LED_BUILTIN, HIGH);
  }
}

void reconnect() {
  while (!client.connected()) {
    if (client.connect("ESP8266-01", mqtt_user, mqtt_pass)) {
      client.subscribe(topic_cmd);
    } else delay(5000);
  }
}

void loop() {
  if (!client.connected()) reconnect();
  client.loop();
  // 每 10 秒上报一次数据
  static unsigned long last = 0;
  if (millis() - last > 10000) {
    float temp = 25.3 + random(-20, 20) / 10.0;
    float hum = 60.0 + random(-50, 50) / 10.0;
    client.publish(topic_temp, String(temp).c_str());
    client.publish(topic_hum, String(hum).c_str());
    last = millis();
  }
}