MQTT协议详解 – 初学者完全指南

​ 目录

1. MQTT协议简介

2. 为什么需要MQTT

3. MQTT基本概念

4. MQTT工作原理

5. MQTT通信模式详解

6. MQTT消息质量(QoS)

7. MQTT主题与通配符

8. MQTT保留消息

9. 遗嘱消息(Last Will and Testament)

10. MQTT会话与清除会话

11. MQTT协议版本对比

12. MQTT安全性考虑

13. MQTT常见应用场景

14. 常见MQTT服务器(Broker)

15. MQTT客户端库

16. MQTT实战示例

17. MQTT与其他协议对比

18. MQTT最佳实践

19. 常见问题与排错

20. 扩展阅读与资源

1. MQTT协议简介

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是一种基于发布/订阅模式的轻量级通信协议，专为受限设备和低带宽、高延迟或不可靠的网络设计。它于1999年由IBM的Andy Stanford-Clark和Arcom的Arlen Nipper开发，最初用于石油和天然气管道的卫星通信。

MQTT协议具有以下特点：

轻量级：最小化网络带宽和设备资源需求
发布/订阅模式：解耦消息发送者和接收者
支持QoS（服务质量）：确保消息可靠传递
保持会话：适用于间歇性连接的设备
双向通信：允许设备上报数据并接收命令
支持最后遗嘱：当客户端异常断开连接时发送预设消息

在物联网(IoT)兴起的背景下，MQTT因其简单高效的特性成为了物联网通信的事实标准之一。

2. 为什么需要MQTT

在传统的HTTP请求-响应模型中，客户端必须主动发起请求才能获取数据，这对于物联网设备存在几个痛点：

能耗高：设备需要定期轮询服务器检查更新
带宽浪费：即使没有新数据，也需要完整的HTTP请求-响应过程
实时性差：数据更新依赖于轮询间隔
部署复杂：设备之间直接通信困难

MQTT协议通过发布/订阅模式解决了这些问题：

低能耗：设备只需订阅一次，不需要定期轮询
带宽优化：协议头部仅2字节起，极为精简
实时性好：有新消息时立即推送
灵活部署：通过中心化broker实现多对多通信
可靠传输：QoS机制确保消息可靠送达，即使在不稳定网络环境

3. MQTT基本概念

3.1 核心角色

MQTT系统包含三个主要角色：

发布者(Publisher)

负责产生消息并发送
可以是传感器、手机应用、服务器等
不关心谁会接收消息

订阅者(Subscriber)

接收特定主题的消息
可以是控制设备、数据分析系统、存储服务等
不知道消息来自哪个发布者

代理服务器(Broker)

MQTT的核心组件，负责消息的接收和分发
管理客户端连接、会话和订阅关系
确保消息按QoS要求可靠传递
实现消息的过滤和路由

3.2 客户端标识(ClientID)

每个连接到MQTT代理的客户端都需要一个唯一的标识符：

用于代理识别客户端
关联到客户端的会话状态
通常由客户端软件自动生成
推荐格式：应用名称+设备ID+时间戳

3.3 连接(Connection)

MQTT建立在TCP/IP协议之上，默认端口为1883(非加密)或8883(加密SSL/TLS)：

客户端通过CONNECT消息与代理建立连接
代理以CONNACK消息响应
连接可以携带认证信息(用户名/密码)
可以设置保持连接(Keep Alive)时间间隔
可以指定清除会话(Clean Session)标志
可以包含遗嘱消息(Will Message)配置

4. MQTT工作原理

4.1 基本通信流程

连接建立：

客户端发送CONNECT消息到Broker
Broker验证客户端信息并返回CONNACK
此时TCP连接已建立并保持

主题订阅：

客户端发送SUBSCRIBE消息，指定感兴趣的主题
Broker确认订阅并返回SUBACK
客户端现在将接收发布到这些主题的消息

消息发布：

发布者发送PUBLISH消息到Broker
Broker接收消息并转发给所有订阅相关主题的客户端
根据QoS级别可能需要确认

连接维护：

客户端定期发送PINGREQ心跳包
Broker回复PINGRESP保持连接活跃
如果超过Keep Alive时间未收到心跳，认为连接断开

断开连接：

客户端可以发送DISCONNECT消息主动断开
如果客户端未正常断开，Broker会发送遗嘱消息

4.2 MQTT控制报文

MQTT定义了14种控制报文，每种报文有特定用途：

每个MQTT报文由三部分组成：

固定头部：所有报文都有，包含报文类型和标志
可变头部：部分报文拥有，包含特定于报文类型的信息
有效载荷：消息的实际内容

5. MQTT通信模式详解

5.1 发布/订阅模式

MQTT采用发布/订阅模式而非传统的请求/响应模式：

发送者和接收者之间完全解耦
数据流通过主题(Topic)进行过滤
一对多广播能力
异步通信，非阻塞

与传统请求/响应模式对比：

5.2 消息过滤方式

MQTT提供基于主题(Topic)的消息过滤，与基于内容过滤相比：

更加轻量，减少计算开销
主题结构可多级设计，形成层次结构
支持通配符订阅，提高灵活性
Broker只需简单字符串匹配即可路由消息

5.3 消息传递保证

MQTT通过QoS机制提供三种级别的消息传递保证，适应不同的应用场景：

QoS 0 (最多一次): 发送后不确认，可能丢失
QoS 1 (至少一次): 确保送达，可能重复
QoS 2 (恰好一次): 确保送达且不重复，有性能开销

5.4 会话管理

MQTT通过会话(Session)概念管理客户端的状态：

包含客户端的订阅列表
保存未确认的QoS 1和QoS 2消息
可以通过Clean Session标志控制持久化
允许客户端断开后重连时恢复状态

6. MQTT消息质量(QoS)

MQTT提供三种QoS级别，平衡可靠性和资源消耗：

6.1 QoS 0 – 最多一次(At most once)

流程：发布者 → Broker → 订阅者，无确认
特点：

最简单且最快的传递方式
无需确认，无状态存储
可能会丢失消息
适合定期发送或非关键数据

典型应用：

传感器定期上报环境数据
状态频繁变化的指标监控
丢失几条消息无关紧要的场景

![QoS 0流程图]

发布者                  Broker                  订阅者
  |                       |                       |
  | --- PUBLISH(QoS 0) -->|                       |
  |                       | --- PUBLISH(QoS 0) -->|
  |                       |                       |

6.2 QoS 1 – 至少一次(At least once)

流程：发布者发送 → Broker确认 → Broker发送 → 订阅者确认
特点：

确保消息至少送达一次
使用PUBACK消息确认
可能导致消息重复
中等开销，需要短期存储状态

典型应用：

计费数据上报
设备控制命令
能接受重复但不能接受丢失的场景

![QoS 1流程图]

发布者                  Broker                  订阅者
  |                       |                       |
  | --- PUBLISH(QoS 1) -->|                       |
  |                       | --- PUBLISH(QoS 1) -->|
  | <---- PUBACK ---------|                       |
  |                       | <---- PUBACK ---------|
  |                       |                       |

6.3 QoS 2 – 恰好一次(Exactly once)

流程：两阶段确认机制，确保消息只传递一次
特点：

最高级别的服务质量
使用四步握手过程(PUBLISH, PUBREC, PUBREL, PUBCOMP)
保证消息不丢失且不重复
较高开销，需要更长状态保持

典型应用：

金融交易记录
关键控制指令
不能容忍丢失或重复的场景

![QoS 2流程图]

发布者                  Broker                  订阅者
  |                       |                       |
  | --- PUBLISH(QoS 2) -->|                       |
  | <---- PUBREC ---------|                       |
  | ----- PUBREL -------->|                       |
  |                       | --- PUBLISH(QoS 2) -->|
  |                       | <---- PUBREC ---------|
  |                       | ----- PUBREL -------->|
  | <---- PUBCOMP --------|                       |
  |                       | <---- PUBCOMP --------|
  |                       |                       |

6.4 QoS降级机制

在MQTT中，QoS可能会发生降级：

Broker向订阅者发送的消息QoS不会超过发布消息的QoS
也不会超过订阅者请求的QoS
最终QoS = min(发布QoS, 订阅QoS)

例如：

如果发布者以QoS 2发布，但订阅者以QoS 1订阅，则消息将以QoS 1传递
如果发布者以QoS 1发布，即使订阅者以QoS 2订阅，消息也只能以QoS 1传递

7. MQTT主题与通配符

7.1 主题格式

MQTT主题是消息路由的基础，主题结构采用层次化设计：

由斜杠(/)分隔的多级结构
大小写敏感
不能包含空格
必须至少包含一个字符
可以包含空主题级别，如”sensor//temperature”

主题样例：

home/livingroom/temperature
vehicles/car1/engine/rpm
user/123/notifications
building/floor5/room503/sensor1

7.2 主题通配符

MQTT提供两种通配符，用于灵活订阅多个主题：

7.2.1 单级通配符(+)

“+”代表单个主题级别的通配符：

只能替代主题结构中的一个级别
可以在主题的任何位置使用

例如：

home/+/temperature 匹配：

home/livingroom/temperature
home/kitchen/temperature

但不匹配：

home/kitchen/upstairs/temperature
home/temperature

7.2.2 多级通配符(#)

“#”代表多个主题级别的通配符：

可以替代零个或多个主题级别
必须位于主题的最后位置
前面必须有斜杠分隔符(/)

例如：

home/# 匹配：

home/livingroom/temperature
home/kitchen/lighting/brightness
home (仅在MQTT 5.0中)

home/+/sensor/# 匹配：

home/livingroom/sensor/temperature
home/kitchen/sensor/humidity/max/min

7.3 主题设计最佳实践

好的主题设计能显著提高系统效率和可维护性：

使用有意义的层次结构

从广泛到具体：区域/设备类型/设备ID/测量类型
例如：building1/floor3/room305/thermostat/temperature

保持一致性

整个系统使用相同的命名约定
同类设备采用相同的主题结构

避免过深的层次结构

通常3-5级是合理的
过深的结构增加复杂性，不易维护

使用设备ID

在主题中包含唯一设备标识符
便于定位和过滤特定设备消息

考虑双向通信

命令和响应使用相关主题
例如：device/123/command 和 device/123/response

预留未来扩展空间

主题结构设计应该考虑未来的扩展需求

8. MQTT保留消息

8.1 保留消息概念

保留消息(Retained Message)是一种特殊的MQTT消息：

标记为”保留”的消息会被Broker存储
当有新客户端订阅匹配主题时，立即收到最新的保留消息
每个主题只能有一条保留消息
发送空有效载荷的保留消息会清除该主题的保留消息

8.2 保留消息的作用

保留消息解决了以下常见问题：

新订阅者不错过最新状态：订阅后立即获知当前状态
离线设备状态同步：设备重连后立即获得最新状态
设备状态持久化：即使设备临时离线，其状态也可被查询
减少请求-响应交互：避免新连接设备需要专门请求状态

8.3 适用场景

保留消息适用于以下场景：

设备状态发布：如开关状态、电池电量、工作模式等
配置信息共享：集中配置信息的分发
“最新值”缓存：如最新传感器读数
设备发现机制：发布设备能力和服务信息

8.4 使用示例

温度传感器状态：

传感器定期发布保留消息到home/livingroom/temperature
任何新应用订阅该主题时立即获得最新温度值
无需等待下一次发布或专门请求

设备在线状态：

设备连接后发布保留消息device/123/status 内容为”online”
当设备正常断开时发送内容为”offline”的保留消息
客户端可立即知道设备当前状态

9. 遗嘱消息(Last Will and Testament)

9.1 遗嘱消息概念

遗嘱消息(LWT)是一种特殊的消息机制：

客户端连接时预先设置的消息
当客户端异常断开时，由Broker代为发布
可以指定主题、QoS级别、保留标志和消息内容
只有在非正常断开连接时才会发送

9.2 触发条件

以下情况会触发遗嘱消息：

网络连接意外断开
客户端响应超时(未在Keep Alive期间发送数据)
客户端未发送DISCONNECT消息就关闭了网络连接
服务器关闭了网络连接(由于协议错误)

以下情况不会触发遗嘱消息：

客户端发送了正常的DISCONNECT消息

9.3 应用场景

遗嘱消息的主要用途：

设备在线状态监控：通知其他设备或系统某设备已离线
资源清理机制：触发清理异常断开设备占用的资源
故障预警：提醒系统管理员设备异常断开
业务连续性保障：触发备用设备接管
会话状态同步：通知相关系统某个会话已结束

9.4 使用示例

在线状态监控：

连接参数：
ClientID: device123
Will Topic: devices/device123/status
Will Message: {"status": "offline", "timestamp": 1620000000}
Will QoS: 1
Will Retain: true

设备连接后发布保留消息：”online”到同一主题
设备异常断开时，Broker自动发布”offline”
监控系统订阅”devices/+/status”即可监控所有设备

异常处理机制：

连接参数：
ClientID: critical-sensor-1
Will Topic: alerts/sensors/failures
Will Message: {"sensorId": "critical-sensor-1", "alert": "connection_lost"}
Will QoS: 2
Will Retain: false

传感器异常断开时触发告警
告警系统收到后启动故障处理流程

10. MQTT会话与清除会话

10.1 会话概念

会话(Session)是客户端与Broker之间的状态信息集合：

包含客户端的所有订阅
包含未确认的QoS 1和QoS 2消息
包含接收但未确认的QoS 2消息
会话可能在客户端断开连接后持续存在

10.2 Clean Session标志

MQTT连接时可以指定Clean Session标志：

Clean Session = true：

创建临时会话，断开连接时销毁
重连时获得全新的会话状态
不会收到离线期间的消息
适合不需要状态持久化的场景

Clean Session = false：

创建持久会话，断开连接时保留
重连时恢复之前的会话状态
可以收到QoS 1和QoS 2的离线消息
适合需要可靠消息传递的场景

10.3 会话状态内容

一个完整的会话状态包括：

客户端的订阅列表
发送给客户端但未确认的QoS 1和QoS 2消息
从客户端接收但未完成确认的QoS 2消息
如果开启了队列机制，还包括排队等待发送的消息

10.4 使用场景对比

11. MQTT协议版本对比

MQTT协议经历了多次演进，主要版本包括：

11.1 MQTT 3.1

最早的标准化版本：

2013年成为OASIS标准
定义了基本的发布/订阅机制
包含QoS、保留消息和遗嘱消息
简单的安全机制(用户名/密码)

11.2 MQTT 3.1.1

最广泛部署的版本：

2014年成为OASIS标准，2016年成为ISO标准(ISO/IEC 20922)
修复了3.1版本的一些问题
改进了协议规范，消除了歧义
增强了错误处理机制
保持向后兼容性

11.3 MQTT 5.0

最新的主要版本，引入大量新特性：

2018年成为OASIS标准
新增功能：

会话/消息过期间隔
原因码和用户属性
共享订阅
主题别名
流量控制
请求/响应模式支持
服务器重定向
增强的认证机制

11.4 主要版本特性对比

12. MQTT安全性考虑

12.1 认证机制

MQTT提供多种认证机制：

客户端ID认证：

最基本的认证方式
仅验证客户端ID的唯一性
安全性极低，不建议作为唯一认证

用户名/密码认证：

CONNECT消息中包含用户名和密码字段
密码在传输中是明文(除非启用TLS)
适合简单应用或结合TLS使用

X.509证书认证(TLS客户端认证)：

使用客户端证书进行双向TLS认证
最安全的认证方式
适合对安全性要求高的应用
需要证书管理基础设施

OAuth/JWT令牌认证：

MQTT 5.0增加了自定义认证支持
可使用JWT作为密码字段
支持基于角色的访问控制

12.2 传输安全

保护MQTT传输层的方法：

TLS/SSL加密：

标准端口1883改为8883(TLS)
加密所有通信内容
防止中间人攻击和窃听
可配置不同安全级别(TLS 1.2, TLS 1.3)

WebSocket+TLS：

用于浏览器环境
通过WSS(WebSocket Secure)传输
标准端口443

VPN通道：

在现有VPN基础设施上运行MQTT
适合企业内部部署

12.3 授权与访问控制

控制客户端权限的方法：

基于主题的访问控制(ACL)：

控制哪些客户端可以发布/订阅哪些主题
支持通配符规则
大多数Broker都提供配置界面

主题命名约定：

将用户/设备ID嵌入主题
例如：users/{userId}/data
结合ACL规则限制只能访问自己的主题

功能分离：

读写分离，使用不同的主题
命令和遥测数据使用不同的主题结构
管理功能使用专门的主题前缀

12.4 安全最佳实践

提高MQTT部署安全性的建议：

始终启用TLS：

即使在内部网络也建议使用
定期更新TLS配置和证书

实施强认证：

避免仅使用客户端ID认证
为每个设备使用唯一凭证
考虑使用证书而非密码

限制连接：

禁用匿名连接
限制每个用户的连接数
实施IP白名单

细粒度授权：

遵循最小权限原则
为不同设备类型创建不同的权限模板
定期审计授权规则

监控异常活动：

记录认证失败
监控异常订阅模式
检测消息速率异常

定期安全审计：

检查过时的证书和凭证
移除未使用的账户
更新安全配置

13. MQTT常见应用场景

13.1 智能家居

MQTT在智能家居中的应用：

设备连接：连接各种家用智能设备
状态同步：同步多个控制界面的状态
传感器监控：收集各种环境传感器数据
场景联动：触发基于多设备状态的自动化
远程控制：实现设备的远程监控和控制

典型主题设计：

home/{location}/{device_type}/{device_id}/{property}
home/{location}/{device_type}/{device_id}/set
home/{location}/{device_type}/{device_id}/status

13.2 工业物联网(IIoT)

工业环境中的MQTT应用：

设备遥测：收集生产设备运行数据
预测性维护：监控设备健康状态
生产监控：实时跟踪生产状态和KPI
能源管理：监控和控制能源使用
告警系统：实时分发关键告警

典型主题设计：

factory/{line_id}/{machine_id}/telemetry
factory/{line_id}/{machine_id}/command
factory/{line_id}/{machine_id}/status
alerts/{severity}/{machine_id}

13.3 车联网

车辆互联应用中的MQTT：

车辆遥测：收集行驶数据和车况
远程诊断：远程检测车辆故障
车队管理：监控和优化车队运营
OTA更新：分发软件更新
车辆控制：远程锁车、空调预热等

典型主题设计：

vehicles/{vehicle_id}/telemetry
vehicles/{vehicle_id}/diagnostics
vehicles/{vehicle_id}/command
vehicles/{vehicle_id}/update

13.4 环境监测

环境监测系统中的MQTT：

传感器网络：汇总分散的环境传感器数据
数据采集：收集温度、湿度、空气质量等数据
异常告警：环境参数超标预警
长期趋势：提供数据用于长期趋势分析

典型主题设计：

sensors/{location}/{sensor_type}/{sensor_id}/data
sensors/{location}/{sensor_type}/{sensor_id}/config
sensors/{location}/{sensor_type}/{sensor_id}/status
alerts/{location}/{parameter}

13.5 医疗健康

MQTT在医疗健康领域的应用：

远程患者监护：收集患者生命体征
医疗设备互联：连接各种医疗设备
药物管理：跟踪药物库存和使用
康复监测：远程监控康复训练
医疗告警：分发紧急医疗事件通知

典型主题设计：

patients/{patient_id}/{device_type}/vitals
patients/{patient_id}/alerts
devices/{department}/{device_id}/status
devices/{department}/{device_id}/telemetry

14. 常见MQTT服务器(Broker)

14.1 Eclipse Mosquitto

简介：轻量级开源MQTT代理
特点：

小巧高效，内存占用低
支持MQTT 3.1, 3.1.1和5.0
支持TLS和WebSockets
适合嵌入式系统和边缘计算

适用场景：

资源受限的环境
小型部署
边缘网关
开发和测试环境

14.2 EMQX

简介：高性能、分布式MQTT消息服务器
特点：

高并发连接(单节点支持百万级连接)
分布式架构，支持水平扩展
丰富的插件生态
完整的企业级功能
支持多协议(MQTT, WebSocket, STOMP等)

适用场景：

大规模物联网部署
需要高可用性的生产环境
需要与其他系统集成的企业环境

14.3 HiveMQ

简介：企业级MQTT消息平台
特点：

高可靠性和可用性
集群支持和水平扩展
企业级安全特性
详细监控和管理工具
云原生设计

适用场景：

企业级IoT部署
关键业务应用
需要专业支持的商业部署

14.4 VerneMQ

简介：高性能、分布式MQTT代理
特点：

基于Erlang/OTP构建
分布式设计
支持多协议
插件扩展系统
集成监控工具

适用场景：

大规模部署
需要高可用性的环境
喜欢Erlang生态的团队

14.5 Mosquitto vs EMQX vs HiveMQ对比

15. MQTT客户端库

15.1 Java客户端

Eclipse Paho Java Client

最流行的Java MQTT客户端
支持MQTT 3.1, 3.1.1和5.0
支持TCP、TLS和WebSocket
多种接口模式：阻塞和异步
官方链接：https://www.eclipse.org/paho/clients/java/

HiveMQ MQTT Client

高性能Java MQTT客户端
支持MQTT 3.1.1和5.0
反应式API
非阻塞I/O
官方链接：https://github.com/hivemq/hivemq-mqtt-client

15.2 Python客户端

Paho-MQTT

最流行的Python MQTT客户端
支持MQTT 3.1和3.1.1
简单易用的API
同步和异步接口
官方链接：https://pypi.org/project/paho-mqtt/

HBMQTT

基于asyncio的MQTT客户端
支持MQTT 3.1.1
异步API设计
包含内置代理
官方链接：https://github.com/beerfactory/hbmqtt

15.3 JavaScript客户端

MQTT.js

流行的Node.js和浏览器MQTT客户端
支持MQTT 3.1和3.1.1
支持WebSocket和TCP
Promise和回调接口
官方链接：https://github.com/mqttjs/MQTT.js

Paho MQTT JavaScript Client

适用于浏览器和Node.js
支持MQTT 3.1和3.1.1
专注于WebSocket传输
官方链接：https://www.eclipse.org/paho/clients/js/

15.4 C/C++客户端

Eclipse Paho C/C++ Client

用于嵌入式和桌面应用
支持MQTT 3.1, 3.1.1和部分5.0
提供同步和异步API
内存占用小
官方链接：https://www.eclipse.org/paho/clients/c/

Mosquitto Client Libraries

Mosquitto项目的一部分
C语言API
简单直接
轻量级
官方链接：https://mosquitto.org/api/

15.5 移动平台客户端

Android – Eclipse Paho Android Service

Android平台专用
提供后台服务管理连接
内存占用优化
官方链接：https://www.eclipse.org/paho/clients/android/

iOS – CocoaMQTT

Swift实现的MQTT客户端
支持MQTT 3.1.1
基于CocoaAsyncSocket
官方链接：https://github.com/emqx/CocoaMQTT

16. MQTT实战示例

16.1 Python示例

16.1.1 基础发布/订阅

import paho.mqtt.client as mqtt
import time

# 连接回调
def on_connect(client, userdata, flags, rc):
    print(f"已连接: {rc}")
    # 订阅主题
    client.subscribe("test/topic")

# 消息接收回调
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.will_set("test/status", "离线", qos=1, retain=True)

# 可选：设置用户名密码
# client.username_pw_set("user", "password")

# 可选：启用TLS
# client.tls_set(ca_certs="ca.crt")

# 连接服务器
client.connect("broker.emqx.io", 1883, 60)

# 发布"在线"状态
client.publish("test/status", "在线", qos=1, retain=True)

# 开始网络循环
client.loop_start()

# 发布10条消息
for i in range(10):
    client.publish("test/topic", f"测试消息 {i}", qos=1)
    time.sleep(1)

# 发布"离线"状态
client.publish("test/status", "离线", qos=1, retain=True)

# 断开连接
client.disconnect()
client.loop_stop()

16.1.2 QoS示例

import paho.mqtt.client as mqtt
import time

def on_connect(client, userdata, flags, rc):
    print(f"已连接: {rc}")

def on_publish(client, userdata, mid):
    print(f"消息已发送: {mid}")

def on_subscribe(client, userdata, mid, granted_qos):
    print(f"已订阅: {mid} QoS: {granted_qos}")

def on_message(client, userdata, msg):
    print(f"收到[QoS {msg.qos}]: {msg.topic} {msg.payload.decode()}")

# 创建客户端
client = mqtt.Client()
client.on_connect = on_connect
client.on_publish = on_publish
client.on_subscribe = on_subscribe
client.on_message = on_message

# 连接服务器
client.connect("broker.emqx.io", 1883, 60)
client.loop_start()

# 订阅不同QoS的主题
client.subscribe([
    ("test/qos0", 0),
    ("test/qos1", 1),
    ("test/qos2", 2)
])

# 使用不同QoS级别发布消息
client.publish("test/qos0", "QoS 0消息", qos=0)
client.publish("test/qos1", "QoS 1消息", qos=1)
client.publish("test/qos2", "QoS 2消息", qos=2)

# 等待消息处理
time.sleep(5)

# 断开连接
client.disconnect()
client.loop_stop()

16.2 JavaScript示例

16.2.1 浏览器客户端

<!DOCTYPE html>
<html>
<head>
    <title>MQTT网页客户端</title>
    <script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
</head>
<body>
    <h1>MQTT测试</h1>
    <div>
        <label>消息:</label>
        <input type="text">

const mqtt = require('mqtt');

// 连接选项
const options = {
    clientId: 'node_client_' + Math.random().toString(16).substring(2, 8),
    clean: true,
    connectTimeout: 4000,
    reconnectPeriod: 1000,
};

// 连接到代理
const client = mqtt.connect('mqtt://broker.emqx.io', options);

// 连接回调
client.on('connect', () => {
    console.log('已连接!');
    
    // 订阅主题
    client.subscribe('test/topic', { qos: 1 }, (err) => {
        if (!err) {
            console.log('已订阅test/topic');
            
            // 发布消息
            setInterval(() => {
                const message = {
                    timestamp: Date.now(),
                    value: Math.random() * 100
                };
                
                client.publish(
                    'test/topic', 
                    JSON.stringify(message), 
                    { qos: 1, retain: false },
                    (err) => {
                        if (!err) {
                            console.log('已发布:', message);
                        }
                    }
                );
            }, 5000);
        }
    });
});

// 消息回调
client.on('message', (topic, message) => {
    console.log(`收到消息: ${topic}: ${message.toString()}`);
});

// 错误处理
client.on('error', (err) => {
    console.error('MQTT错误:', err);
});

// 断线处理
client.on('close', () => {
    console.log('连接已关闭');
});

// 重连处理
client.on('reconnect', () => {
    console.log('正在重连...');
});

// 处理退出
process.on('SIGINT', () => {
    client.end(true, () => {
        console.log('已安全断开连接');
        process.exit(0);
    });
});

#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>

// WiFi设置
const char* ssid = "YourWiFiSSID";
const char* password = "YourWiFiPassword";

// MQTT设置
const char* mqtt_server = "broker.emqx.io";
const int mqtt_port = 1883;
const char* mqtt_client_id = "arduino_sensor";
const char* mqtt_topic_temp = "home/livingroom/temperature";
const char* mqtt_topic_humidity = "home/livingroom/humidity";
const char* mqtt_topic_status = "home/livingroom/sensor/status";

// DHT传感器设置
#define DHTPIN 2     // 传感器连接到引脚2
#define DHTTYPE DHT22   // DHT22传感器
DHT dht(DHTPIN, DHTTYPE);

WiFiClient espClient;
PubSubClient client(espClient);
unsigned long lastMsg = 0;
char tempMsg[50];
char humMsg[50];

void setup_wifi() {
  delay(10);
  Serial.println();
  Serial.print("连接到WiFi: ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  Serial.println("");
  Serial.println("WiFi已连接");
  Serial.println("IP地址: ");
  Serial.println(WiFi.localIP());
}

void reconnect() {
  while (!client.connected()) {
    Serial.print("尝试MQTT连接...");
    if (client.connect(mqtt_client_id, NULL, NULL, mqtt_topic_status, 1, true, "offline")) {
      Serial.println("已连接");
      // 发布上线状态
      client.publish(mqtt_topic_status, "online", true);
    } else {
      Serial.print("失败, rc=");
      Serial.print(client.state());
      Serial.println(" 5秒后重试");
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200);
  Serial.println("DHT22传感器测试");
  
  dht.begin();
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  unsigned long now = millis();
  if (now - lastMsg > 30000) {  // 每30秒发送一次
    lastMsg = now;
    
    // 读取温湿度
    float h = dht.readHumidity();
    float t = dht.readTemperature();

    // 检查读取是否成功
    if (isnan(h) || isnan(t)) {
      Serial.println("无法读取DHT传感器!");
      return;
    }

    // 发布温度
    snprintf(tempMsg, 50, "%.1f", t);
    Serial.print("温度: ");
    Serial.println(tempMsg);
    client.publish(mqtt_topic_temp, tempMsg, true);  // 保留消息
    
    // 发布湿度
    snprintf(humMsg, 50, "%.1f", h);
    Serial.print("湿度: ");
    Serial.println(humMsg);
    client.publish(mqtt_topic_humidity, humMsg, true);  // 保留消息
  }
}