MQTT 协议模型：客户端、 broker 与主题详解（二）

 三、MQTT 客户端

3.1 客户端定义

MQTT 客户端是指运行 MQTT 库并通过网络连接到 MQTT broker 的任何设备 ，这些设备可以是从资源极其有限的微控制器，到功能强大的成熟服务器等各种类型。在 MQTT 的发布 / 订阅模型中，客户端扮演着发布者（Publisher）和订阅者（Subscriber）的角色。发布者负责向 broker 发送消息，订阅者则从 broker 接收感兴趣的消息 。例如，在一个智能农业系统中，土壤湿度传感器作为 MQTT 客户端，可以作为发布者，定时将采集到的土壤湿度数据发布到 “agriculture/soil/moisture” 主题；而农场管理人员的监控终端作为另一个 MQTT 客户端，作为订阅者，订阅该主题，从而实时获取土壤湿度信息，以便做出相应的灌溉决策 。

3.2 生命周期与关键操作

3.2.1 建立连接

客户端在连接 MQTT broker 时，需要配置一系列关键参数：

• 接入地址和端口：指定 MQTT broker 的网络地址（可以是 IP 地址或域名）以及端口号，默认情况下，MQTT 使用 TCP 端口 1883 进行通信，如果需要使用 TLS 加密通信，则通常使用端口 8883 。例如，若 broker 的地址为 “broker.example.com”，使用默认端口，那么接入地址和端口配置为 “broker.example.com:1883” 。

• 传输类型：常见的传输类型有 TCP、WebSocket 等。TCP 是最常用的传输方式，提供可靠的字节流传输；WebSocket 则适用于需要在 Web 浏览器中使用 MQTT 的场景，它可以在 HTTP 协议之上建立 MQTT 连接，突破浏览器的同源策略限制 。

• 认证信息：当 broker 启用了认证鉴权机制时，客户端需要提供对应的用户名（Username）和密码（Password）进行身份验证，以确保只有合法的客户端能够连接到 broker 。

• 其他参数：还包括设置心跳时间间隔（Keep Alive Interval），用于指定客户端向 broker 发送心跳包的时间周期，以保持连接的活性；以及设置清理会话标志（Clean Session），若设置为 true，客户端断开连接时，broker 不会保留其会话状态，包括订阅关系和未处理的消息等；若为 false，broker 会保留会话状态，客户端重新连接时可以恢复之前的会话 。例如在一个智能家居项目中，智能摄像头客户端连接到 MQTT broker 时，配置如下：

import paho.mqtt.client as mqtt

# 创建MQTT客户端实例

client = mqtt.Client(client_id="smart_camera_001", clean_session=True)

# 设置用户名和密码

client.username_pw_set(username="camera_user", password="123456")

# 设置心跳时间间隔为60秒

client.keep_alive = 60

# 连接到MQTT broker

client.connect("mqtt.example.com", port=1883)

3.2.2 订阅主题

客户端通过订阅主题来接收感兴趣的消息。在订阅主题时，需要指定主题过滤器（Topic Filter），主题过滤器可以包含通配符，用于匹配多个主题。例如，“home/+/temperature” 中的 “+” 是单级通配符，表示可以匹配 “home” 下任意一级的 “temperature” 主题，如 “home/livingroom/temperature”“home/bedroom/temperature” 等 ；“home/#” 中的 “#” 是多级通配符，表示可以匹配 “home” 下的所有主题 。同时，还需要指定服务质量（QoS）级别，QoS 定义了消息传输的可靠性级别，取值有 0、1、2 三种 ：

• QoS 0：最多一次，消息发布后不等待确认，可能会丢失，适用于对消息可靠性要求不高的场景，如实时性要求较高但允许少量数据丢失的环境监测数据采集 。

• QoS 1：至少一次，消息会被发送，并且会等待 broker 的确认，如果未收到确认则会重发，可能会出现重复消息，适合大部分物联网场景，如设备状态上报 。

• QoS 2：恰好一次，确保消息只被送达一次，不会重复也不会丢失，通过复杂的握手机制实现，常用于对数据准确性要求极高的场景，如金融交易数据传输 。

以 Python 的 paho - mqtt 库为例，订阅主题的代码如下：

def on_connect(client, userdata, flags, rc):

if rc == 0:

print("Connected to MQTT Broker!")

# 订阅主题，QoS设置为1

client.subscribe("home/+/temperature", qos=1)

else:

print(f"Failed to connect, return code {rc}")

client.on_connect = on_connect

3.2.3 发布消息

客户端向指定主题发布消息时，需要设置以下参数：

• 目标主题（Topic）：明确消息要发布到的主题，例如 “office/lighting/control”，用于标识消息的类别和目标受众 。

• 消息内容（Payload）：即真正要传输的应用层数据，可以是文本、二进制数据、JSON 格式数据等，具体内容和格式由应用层决定。例如，智能家居系统中，控制智能灯泡开关的消息内容可以是 “ON” 或 “OFF” 的字符串；而传感器上报的数据可能是 JSON 格式，如{"temperature": 25.5, "humidity": 60} 。

• 服务质量（QoS）：与订阅主题时的 QoS 含义相同，指定消息传递的可靠性级别 。

• 保留标志（Retain）：可选参数，若设置为 true，broker 会存储这条消息，当有新的客户端订阅该主题时，会立即收到这条保留的消息，适用于发布设备的当前状态等重要信息，使新订阅者能快速获取最新状态 。

使用 paho - mqtt 库发布消息的示例代码：

# 发布消息，QoS为1，不保留消息

message = "Turn on the lights"

client.publish("office/lighting/control", payload=message, qos=1, retain=False)

3.2.4 取消订阅与断开连接

• 取消订阅：当客户端不再需要接收某个主题的消息时，可以向 broker 发送取消订阅请求（UNSUBSCRIBE），指定要取消订阅的主题。例如在智能家居系统中，当用户不再关注某个房间的温度数据时，对应的客户端可以取消对 “home/bedroom/temperature” 主题的订阅 。取消订阅操作可以减少客户端的资源消耗，同时也减轻 broker 的消息转发负担 。

• 断开连接：客户端完成通信任务后，可以主动向 broker 发送断开连接请求（DISCONNECT），正常关闭与 broker 的网络连接。在断开连接前，客户端可以处理一些清理工作，如保存未处理的消息等 。此外，如果客户端意外断开连接（如设备断电、网络故障等），broker 会根据客户端连接时设置的遗嘱消息（Last Will Message，LWT）进行处理，若设置了遗嘱消息，broker 会将遗嘱消息发布到指定主题，告知其他订阅者客户端的异常断开情况 。例如在工业监控系统中，某台设备客户端意外断电，broker 会将其遗嘱消息 “device X has disconnected unexpectedly” 发布到 “industrial/device/status” 主题，方便监控人员及时知晓设备状态 。

3.3 发布者与订阅者

3.3.1 发布者工作原理

发布者是 MQTT 客户端中负责向 broker 发送消息的角色。当发布者连接到 broker 之后，就可以构建并发送 PUBLISH 消息 。每条 PUBLISH 消息都包含以下关键部分：

• 主题（Topic）：明确消息所属的类别和目标受众，例如 “weather/station1/temperature”，它是 broker 进行消息路由的重要依据 。

• 消息负载（Payload）：即真正发布的应用消息内容，其格式和含义由应用层定义。如气象监测系统中，发布的温度数据可能是 “28.5” 这样的浮点数；智能家居中，智能门锁的状态消息负载可能是 JSON 格式{"status": "locked", "timestamp": "2024-10-01 12:00:00"} 。

• 服务质量（QoS）标志：标记当前消息传送的交付水平，取值为 QoS 0（最多发送一次）、QoS 1（最少发送一次）、QoS 2（正好发送一次）。不同的 QoS 级别决定了消息传输的可靠性和开销，QoS 0 传输速度快但可能丢失消息；QoS 1 通过确认机制保证消息至少被接收一次，但可能重复；QoS 2 通过复杂的四次握手确保消息只被接收一次，可靠性最高但开销也最大 。

• 保留标志（Retain）：当 Retain = 1 时，broker 会存储该消息，当新的 client 注册订阅且匹配该消息的主题时，该保留的消息会发送给订阅者，常用于发布设备的最新状态等重要信息 。

• 重传标志（DUP）：标记当前的 PUBLISH/PUBREL 消息是否为重复发送消息，当网络不稳定导致发送方未收到接收方的确认消息时，可能会重发消息，此时 DUP 标志会被设置 。

例如，在一个智能交通系统中，车辆作为发布者，向 “traffic/vehicle/status” 主题发布自身的行驶速度、位置等信息：

import paho.mqtt.client as mqtt

# 创建MQTT客户端实例

client = mqtt.Client()

# 连接到MQTT broker

client.connect("mqtt.example.com", 1883)

# 构建消息内容

message_payload = '{"speed": 60, "latitude": 39.9, "longitude": 116.3}'

# 发布消息，QoS为1，不保留消息

client.publish("traffic/vehicle/status", payload=message_payload, qos=1, retain=False)

# 断开连接

client.disconnect()

3.3.2 订阅者工作原理

订阅者是 MQTT 客户端中负责接收感兴趣主题消息的角色。订阅者在与 broker 建立连接后，通过向 broker 发送 SUBSCRIBE 消息来订阅感兴趣的主题 。SUBSCRIBE 消息的有效载荷中至少需要包含一个主题过滤器（由 topic 和 QoS 组成） 。例如，在一个城市环境监测系统中，数据处理中心作为订阅者，订阅 “city/environment/pm25”“city/environment/temperature” 等主题，以获取空气质量和温度数据 。

当 broker 接收到订阅者的 SUBSCRIBE 消息后，会返回 SUBACK 消息作为订阅确认 。SUBACK 消息包含一个返回码列表，用于告知订阅者每个主题订阅的结果，返回码 0x00 表示订阅成功，最大 QoS 为 0；0x01 表示订阅成功，最大 QoS 为 1；0x02 表示订阅成功，最大 QoS 为 2；0x80 表示订阅失败 。

一旦订阅成功，当有发布者向订阅者所订阅的主题发布消息时，broker 会将消息转发给订阅者 。订阅者通过设置消息回调函数来处理接收到的消息，在回调函数中可以对消息进行解析、存储、处理等操作 。例如，使用 paho - mqtt 库的 Python 代码如下：

def on_connect(client, userdata, flags, rc):

if rc == 0:

print("Connected to MQTT Broker!")

# 订阅主题，QoS设置为1

client.subscribe("city/environment/pm25", qos=1)

client.subscribe("city/environment/temperature", qos=1)

else:

print(f"Failed to connect, return code {rc}")

def on_message(client, userdata, msg):

print(f"Received message on topic '{msg.topic}': {msg.payload.decode()}")

client = mqtt.Client()

client.on_connect = on_connect

client.on_message = on_message

client.connect("mqtt.example.com", 1883)

client.loop_start()

如果订阅者不再对某些主题感兴趣，可以向 broker 发送 UNSUBSCRIBE 消息取消订阅，UNSUBSCRIBE 消息中至少需要包含一个要取消订阅的主题 。broker 接收到 UNSUBSCRIBE 消息后，会返回 UNSUBACK 消息作为取消订阅的确认 。

四、MQTT Broker

4.1 Broker 定义与作用

MQTT Broker 是 MQTT 通信架构中的核心组件，充当消息代理的角色 。它就像是一个智能的邮件中转站，负责接收来自各个客户端的消息，然后根据订阅关系将这些消息准确无误地分发给相应的订阅者客户端 。例如在一个大型的智能工厂中，有数百个传感器作为客户端，它们不断地采集设备运行数据（如温度、压力、转速等），并将这些数据发布到 MQTT Broker。而工厂的监控系统作为订阅者客户端，通过订阅相关主题，就能实时获取这些传感器数据，实现对生产过程的实时监控和分析 。

具体来说，MQTT Broker 的核心作用包括：

• 消息转发：这是 Broker 最主要的功能。当发布者客户端向某个主题发布消息时，Broker 会接收该消息，并根据其内部维护的主题订阅关系表，将消息转发给所有订阅了该主题的客户端 。例如在智能家居系统中，智能摄像头作为发布者，将拍摄到的监控画面信息发布到 “home/security/camera” 主题，而用户的手机 APP 作为订阅者，只要订阅了该主题，就能通过 Broker 接收并查看监控画面 。

• 主题管理：Broker 负责管理系统中的所有主题，包括主题的创建（虽然主题通常由客户端在发布或订阅时隐式创建，但 Broker 需要维护主题相关信息）、删除（当没有客户端订阅某个主题且无保留消息等情况时，Broker 可能会清理相关主题资源）以及主题权限控制（在一些安全要求较高的场景中，Broker 会根据配置对不同客户端对主题的发布和订阅权限进行管理） 。

• 会话管理：对于设置了清理会话标志为 false 的客户端连接，Broker 会维护其会话状态 。这包括记录客户端的订阅关系，以及在客户端离线期间存储发送给该客户端但尚未送达的消息（根据 QoS 级别而定） 。当客户端重新连接时，Broker 能够恢复其会话状态，将未处理的消息发送给客户端 。例如在车联网场景中，车辆客户端在行驶过程中可能会因为信号问题短暂离线，当信号恢复重新连接到 Broker 时，Broker 会将离线期间积累的交通路况更新等消息发送给车辆客户端 。

4.2 常见 Broker 介绍

• EMQX：EMQX 是一款基于 Erlang/OTP 平台开发的开源 MQTT Broker，它在性能和功能上表现卓越 。其高性能体现在能够支持海量的并发连接，单节点可轻松支持百万级别的 TCP 连接 ，适用于大规模物联网设备接入的场景，如智能城市中的各类传感器、智能交通中的车辆等设备连接 。功能方面，EMQX 提供了丰富的插件体系，通过插件可以实现与多种数据库（如 MySQL、Redis 等）的集成，方便数据的存储和查询；支持 WebSocket 协议，便于在 Web 应用中使用 MQTT 通信；还具备强大的安全认证和访问控制功能，包括 TLS 加密、用户名密码认证、客户端证书认证等，保障通信的安全性 。例如在智能电网项目中，EMQX 可以作为 MQTT Broker，连接分布在各个区域的电力监测设备，实现电力数据的实时采集和分析，通过其与数据库的集成功能，将历史电力数据存储到 MySQL 数据库中，供后续的数据分析和报表生成使用 。

• Mosquitto：Mosquitto 是另一个广受欢迎的开源 MQTT Broker，由 Eclipse 基金会维护 。它以轻量级和易于部署著称，非常适合在资源受限的环境中运行，如小型物联网网关、树莓派等设备 。Mosquitto 遵循 MQTT 协议标准，提供了基本的消息代理功能，包括消息转发、主题管理和会话管理等 。它支持多种操作系统，如 Linux、Windows、macOS 等，方便开发者在不同平台上进行开发和部署 。例如在一个基于树莓派搭建的智能家居实验系统中，Mosquitto 可以运行在树莓派上，作为本地的 MQTT Broker，连接智能灯泡、智能插座等设备，实现简单的智能家居控制功能 。同时，Mosquitto 还提供了命令行工具，方便用户进行测试和调试，如通过 mosquitto_pub 命令发布消息，mosquitto_sub 命令订阅消息 。

• HiveMQ：HiveMQ 是一款企业级的 MQTT Broker，专注于提供高性能、高可靠性和安全性的 MQTT 服务 。它支持集群部署，通过集群可以实现负载均衡和故障转移，确保系统的高可用性，适用于对系统稳定性要求极高的企业级物联网应用，如工业自动化生产线监控、金融物联网中的交易设备通信等场景 。HiveMQ 提供了丰富的企业级功能，包括高级安全特性，如基于角色的访问控制（RBAC），可以精细地控制不同用户或设备对 MQTT 资源的访问权限；支持与企业现有的安全基础设施集成，如与 LDAP（轻量级目录访问协议）服务器集成，实现用户身份验证和授权 。此外，HiveMQ 还提供了 REST API，方便与其他系统进行集成和管理，例如企业可以通过 REST API 获取 MQTT Broker 的运行状态、统计信息等，实现对 MQTT 服务的集中监控和管理 。

4.3 Broker 工作流程

4.3.1 接收连接请求

当 MQTT 客户端向 Broker 发起连接请求时，Broker 会进行一系列处理：

1. 建立 TCP 连接：客户端首先通过 TCP 协议与 Broker 建立网络连接，这是 MQTT 通信的基础 。在这个过程中，客户端需要指定 Broker 的 IP 地址（或域名）和端口号，默认情况下，MQTT 使用 TCP 端口 1883 进行通信，如果需要使用 TLS 加密通信，则通常使用端口 8883 。例如，在一个智能农业项目中，土壤湿度传感器客户端连接到位于云端的 MQTT Broker 时，会通过 TCP 协议尝试连接到 “broker.example.com:1883” 。

1. 解析 CONNECT 报文：连接建立后，客户端会向 Broker 发送 CONNECT 报文，报文中包含了客户端的基本信息，如客户端 ID（Client ID）、用户名（Username）、密码（Password）、遗嘱消息（Last Will Message，LWT）、心跳时间间隔（Keep Alive Interval）以及清理会话标志（Clean Session）等 。Broker 会解析这些信息：

• • 验证客户端 ID：客户端 ID 用于唯一标识一个客户端，Broker 会检查客户端 ID 是否合法以及是否与已连接的客户端 ID 冲突 。如果客户端 ID 不合法或已被占用，Broker 可能会拒绝连接 。例如，在一个车联网应用中，每辆车的车载终端作为 MQTT 客户端，其客户端 ID 可以设置为车辆的唯一识别码（VIN），确保每个客户端 ID 的唯一性 。

• • 认证与授权：如果客户端在 CONNECT 报文中提供了用户名和密码，Broker 会根据配置的认证策略进行身份验证，验证通过后才允许客户端连接 。一些高级的 Broker 还支持基于角色的访问控制（RBAC）等授权机制，根据客户端的角色或权限，限制其对 MQTT 资源（如主题的发布和订阅权限）的访问 。

• • 处理遗嘱消息：如果客户端设置了遗嘱消息，Broker 会记录相关信息，包括遗嘱消息的主题、内容和 QoS 等级 。当客户端意外断开连接（如设备断电、网络故障等，而不是正常发送 DISCONNECT 报文断开连接）时，Broker 会将遗嘱消息发布到指定主题 。例如在工业监控系统中，某台设备客户端设置遗嘱消息为 “device/1/offline”，当该设备意外离线时，Broker 会将此遗嘱消息发布到相应主题，通知其他订阅者该设备已离线 。

• • 会话管理：根据客户端 CONNECT 报文中的清理会话标志（Clean Session），Broker 决定是否保留客户端的会话状态 。如果 Clean Session 为 true，客户端断开连接时，Broker 不会保留其会话状态，包括订阅关系和未处理的消息等；如果为 false，Broker 会保留会话状态，客户端重新连接时可以恢复之前的会话 。

1. 返回 CONNACK 报文：Broker 完成对 CONNECT 报文的处理后，会向客户端返回 CONNACK 报文，报文中包含连接结果码 。连接结果码为 0x00 表示连接成功；其他值表示不同的连接失败原因，如 0x01 表示协议版本不匹配，0x04 表示用户名或密码错误等 。客户端根据 CONNACK 报文中的连接结果码，判断连接是否成功，若失败则进行相应处理，如提示用户或尝试重新连接 。

4.3.2 处理消息发布与订阅

1. 处理订阅请求：当客户端向 Broker 发送 SUBSCRIBE 报文进行主题订阅时，Broker 会执行以下操作：

• • 解析 SUBSCRIBE 报文：Broker 解析 SUBSCRIBE 报文，获取客户端订阅的主题过滤器（Topic Filter）和期望的服务质量等级（QoS） 。主题过滤器可以包含通配符，用于匹配多个主题 。例如，“home/+/temperature” 中的 “+” 是单级通配符，表示可以匹配 “home” 下任意一级的 “temperature” 主题 。

• • 验证与记录订阅信息：Broker 验证订阅请求的合法性，如主题过滤器是否符合语法规范 。验证通过后，Broker 会将客户端的订阅信息记录到其内部维护的订阅关系表中，该表记录了每个主题与订阅该主题的客户端列表以及对应的 QoS 等级 。

• • 返回 SUBACK 报文：Broker 向客户端返回 SUBACK 报文，确认订阅请求 。SUBACK 报文中包含一个返回码列表，每个返回码对应一个订阅的主题过滤器，用于告知客户端每个主题订阅的结果 。返回码 0x00 表示订阅成功，最大 QoS 为 0；0x01 表示订阅成功，最大 QoS 为 1；0x02 表示订阅成功，最大 QoS 为 2；0x80 表示订阅失败 。

1. 处理发布请求：当客户端向 Broker 发送 PUBLISH 报文发布消息时，Broker 会进行如下处理：

• • 解析 PUBLISH 报文：Broker 解析 PUBLISH 报文，获取消息的主题、消息内容（Payload）、服务质量等级（QoS）和保留标志（Retain）等信息 。

• • 消息路由与分发：Broker 根据消息的主题，在其订阅关系表中查找订阅了该主题的客户端列表 。然后，根据每个客户端订阅时指定的 QoS 等级，将消息转发给相应的客户端 。在转发过程中，如果客户端当前处于离线状态，对于 QoS 1 和 QoS 2 的消息，Broker 会根据会话管理策略，存储消息以便在客户端重新连接时发送；对于 QoS 0 的消息，由于其 “至多一次” 的特性，通常不会存储 。例如，在一个智能物流系统中，货物运输车辆客户端向 “logistics/vehicle/status” 主题发布车辆位置、行驶状态等消息，Broker 会将这些消息分发给订阅了该主题的物流监控中心客户端 。

• • 处理保留消息：如果 PUBLISH 报文中的保留标志（Retain）为 1，Broker 会存储该消息 。当有新的客户端订阅该主题时，Broker 会将这条保留消息发送给新订阅者 。每个主题只能有一条保留消息，新的保留消息会覆盖旧的 。例如在智能家居系统中，智能门锁客户端发布当前门锁状态（“locked” 或 “unlocked”）的消息并设置为保留消息，新连接的手机 APP 客户端订阅该主题时，能立即获取到门锁的当前状态 。

4.3.3 管理会话与心跳机制

1. 会话管理：

• • 会话建立：当客户端与 Broker 建立连接时，如果 CONNECT 报文中的清理会话标志（Clean Session）为 false，Broker 会为客户端创建一个会话，并将会话相关信息（如订阅关系、未发送的消息等）存储在内存或持久化存储中（取决于 Broker 的配置） 。例如在一个远程医疗监测系统中，患者佩戴的可穿戴设备客户端与 Broker 建立连接时设置 Clean Session 为 false，Broker 会为其创建会话，记录设备订阅的用于接收医生指令的主题以及设备发布健康数据的主题等信息 。

• • 会话维持：在会话期间，Broker 会维护客户端的订阅关系和未处理的消息 。如果客户端向 Broker 发送新的订阅、取消订阅或发布消息等请求，Broker 会相应地更新会话信息 。

• • 会话结束：当客户端正常发送 DISCONNECT 报文断开连接时，Broker 会根据会话设置和消息处理情况，清理会话相关资源 。如果客户端意外断开连接，Broker 会根据会话设置和遗嘱消息等机制进行处理 。例如，如果客户端设置了遗嘱消息，Broker 会在检测到客户端意外断开后，发布遗嘱消息；对于设置了 Clean Session 为 false 的客户端，Broker 会保留一定时间的会话信息，等待客户端重新连接 。

1. 心跳机制：

• • 心跳包发送：客户端在与 Broker 建立连接时，会在 CONNECT 报文中指定心跳时间间隔（Keep Alive Interval），单位为秒 。客户端需要每隔 Keep Alive Interval 的一半时间向 Broker 发送 PINGREQ 报文，以表明自己仍然在线 。例如，如果客户端设置的心跳时间间隔为 60 秒，那么它需要每隔 30 秒向 Broker 发送一次 PINGREQ 报文 。

• • 心跳响应处理：Broker 收到客户端的 PINGREQ 报文后，会立即返回 PINGRESP 报文作为响应 。如果 Broker 在 Keep Alive Interval 时间内没有收到客户端的 PINGREQ 报文，且又没有其他数据通信，Broker 会认为客户端可能已经离线，会根据会话设置和遗嘱消息机制进行相应处理，如发布遗嘱消息、清理会话资源（对于长时间未连接的客户端）等 。

• • 心跳机制的作用：心跳机制主要用于检测客户端与 Broker 之间的网络连接状态，确保连接的有效性 。在物联网环境中，网络状况可能不稳定，设备可能会因为各种原因（如信号干扰、电源故障等）意外断开连接 。通过心跳机制，Broker 能够及时发现这些异常情况，从而采取相应措施，保障 MQTT 通信系统的可靠性和稳定性 。