Python实现MQTT发布订阅原理与代码解析

木白

发布时间：2026-10-14 | 4人浏览过

原创

本文深入解析基于Python的MQTT发布订阅机制，涵盖协议核心特点、QoS服务质量等级及客户端服务端交互原理。通过提供完整的Python客户端发布与服务端订阅代码示例，展示如何在物联网场景中利用轻量级MQTT协议实现设备间的高效通信与数据同步。

MQTT协议简介与特点

MQTT简介

MQTT全称为Message Queuing Telemetry Transport（消息队列遥测传输），是一种基于发布/订阅范式的轻量级消息协议，构建于TCP/IP协议栈之上。

该协议具备轻量、简单、开放及易于实现的特点，适用范围极广。尤其在受限环境中，如机器对机器（M2M）通信和物联网（IoT）领域，其优势尤为显著。

目前，MQTT已广泛应用于卫星链路通信传感器、间歇性拨号的医疗设备、智能家居以及各类小型化设备中。

MQTT特点

1、采用发布/订阅消息模式，支持一对多的消息发布，有效解除应用程序间的耦合。协议涉及客户端和服务端三种主要身份：发布者（Publisher）、代理（Broker，即服务器）和订阅者（Subscriber）。其中，消息的发布者和订阅者均为客户端，消息代理作为服务器运行，且消息发布者可同时充当订阅者，从而实现了生产者与消费者的解耦。

2、对负载内容实施屏蔽的消息传输机制。

3、依赖TCP/IP协议提供网络连接支持。

4、提供三种消息发布服务质量（QoS）等级：

“至多一次”：消息发布完全依赖底层TCP/IP网络，可能导致消息丢失或重复。适用于环境传感器数据等场景，偶尔丢失一次记录影响不大，因为后续会有新的数据发送。
“至少一次”：确保消息能够到达目的地，但允许消息重复发生。
“只有一次”：确保消息仅被送达一次。适用于计费系统等场景，因为消息的重复或丢失会导致结果错误。

5、传输开销极小，固定长度头部仅为2字节，最小化协议交换以降低网络流量。

6、利用Last Will和Testament（遗嘱）特性，通知相关各方客户端异常中断的情况。

MQTT工作原理与核心概念

原理

MQTT协议包含三种身份：发布者、代理和订阅者。发布者和订阅者均属于客户端范畴，代理则为服务器。通常情况下，为了节省内存和流量，消息发布者与订阅者会被定义在同一实体中。

MQTT传输的消息由主题（Topic，可视为消息类型，订阅者订阅后接收该主题下的消息内容payload）和负载（payload，即消息的具体内容）两部分组成。

MQTT协议实现方式详解

1.MQTT协议实现方式

实现MQTT协议需客户端与服务器端完成通讯。在此过程中，MQTT协议定义了三种身份：发布者（Publish）、代理（Broker，服务器）和订阅者（Subscribe）。消息的发布者和订阅者均为客户端，消息代理作为服务器运行，且消息发布者可以同时是订阅者。

MQTT传输的消息主要包含两部分：主题（Topic）和负载（payload）。

（1）Topic：可理解为消息的类型。订阅者订阅后，将收到该主题下的消息内容（payload）。

（2）payload：可理解为消息的具体内容，即订阅者实际处理的数据。

2.网络传输与应用消息

MQTT负责构建底层网络传输，建立客户端到服务器的连接，提供有序、无损且基于字节流的双向传输通道。

当应用数据通过MQTT网络发送时，MQTT会将服务质量（QoS）和主题名（Topic）关联起来。

3.MQTT客户端

使用MQTT协议的应用程序或设备始终建立到服务器的网络连接。客户端具备以下功能：

（1）发布其他客户端可能订阅的信息；

（2）订阅其他客户端发布的消息；

（3）退订或删除应用程序的消息；

（4）断开与服务器的连接。

4.MQTT服务端

MQTT服务器被称为“消息代理”（Broker），可以是应用程序或设备。它位于消息发布者和订阅者之间，具备以下功能：

（1）接受来自客户的网络连接；

（2）接受客户发布的应用信息；

（3）处理来自客户端的订阅和退订请求；

（4）向订阅的客户转发应用程序消息。

5.MQTT协议中的订阅、主题、会话

一、订阅（Subscription）

订阅包含主题筛选器（Topic Filter）和最大服务质量（QoS）。订阅与一个会话（Session）关联，一个会话可包含多个订阅。每个会话中的每个订阅必须拥有不同的主题筛选器。

二、会话（Session）

每个客户端与服务器建立连接后即形成一个会话，客户端和服务器之间存在状态交互。会话存在于一个网络中，也可能在客户端和服务器之间跨越多个连续的网络连接。

三、主题名（Topic Name）

连接应用程序消息的标签，该标签与服务器上的订阅相匹配。服务器会将消息发送给订阅了匹配标签的每个客户端。

四、主题筛选器（Topic Filter）

一种对主题名使用通配符的筛选器，用于订阅表达式中，表示订阅所匹配的多个主题。

五、负载（Payload）

消息订阅者具体接收的内容。

MQTT协议定义的方法

6.MQTT协议中的方法

MQTT协议定义了一些方法（也称为动作），用于对特定资源进行操作。该资源可以是预先存在的数据或动态生成的数据，具体取决于服务器实现。通常，资源指服务器上的文件或输出。主要方法包括：

（1）Connect：等待与服务器建立连接。

（2）Disconnect：等待MQTT客户端完成所做的工作，并与服务器断开TCP/IP会话。

（3）Subscribe：等待完成订阅。

（4）UnSubscribe：等待服务器取消客户端的一个或多个topics订阅。

（5）Publish：MQTT客户端发送消息请求，发送完成后返回应用程序线程。

MQTT的应用场景

7.应用场景

由于MQTT传输消息具有异步性（发布订阅模式），且协议本身轻量，因此适用于轻量级应用。

它可作为物联网的通信组件，例如在树莓派上搭建MQTT服务器。随着智能家居的全面普及，家居中的消息通讯均可通过此协议实现。

例如智能冰箱温度检测、房间温度检测等信息均可通过MQTT实现。此外，遥感数据、汽车检测数据、智能家居、智慧城市及医疗医护等领域均具有广泛的应用场景。

Python实现MQTT客户端代码

客户端


#!/usr/bin/env python 
# encoding: utf-8 
""" 
@version: v1.0 
@author: W_H_J 
@license: Apache Licence 
@contact: 415900617@qq.com 
@software: PyCharm 
@file: clicentMqttTest.py 
@time: 2019/2/22 14:19 
@describe: mqtt客户端
"""
import json
import sys
import os
import paho.mqtt.client as mqtt
import time
 
sys.path.append(os.path.abspath(os.path.dirname(__file__) + '/' + '..'))
sys.path.append("..")
 
TASK_TOPIC = 'test' # 客户端发布消息主题
 
client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
"""
client_id是连接到代理。如果client_id的长度为零或为零，则行为为由使用的协议版本定义。如果使用MQTT v3.1.1，
那么一个零长度的客户机id将被发送到代理，代理将被发送为客户端生成一个随机变量。如果使用MQTT v3.1，那么id将是
随机生成的。在这两种情况下，clean_session都必须为True。如果这在这种情况下不会产生ValueError。
注意：一般情况下如果客户端服务端启用两个监听那么客户端client_id 不能与服务器相同，如这里用时间"20190222142358"作为它的id，
如果与服务器id相同，则无法接收到消息
"""
client = mqtt.Client(client_id, transport='tcp')
 
client.connect("127.0.0.1", 1883, 60) # 此处端口默认为1883，通信端口期keepalive默认60
client.loop_start()
 
 
def clicent_main(message: str):
  """
  客户端发布消息
  :param message: 消息主体
  :return:
  """
  time_now = time.strftime('%Y-%m-%d %H-%M-%S', time.localtime(time.time()))
  payload = {"msg": "%s" % message, "data": "%s" % time_now}
  # publish(主题：Topic; 消息内容)
  client.publish(TASK_TOPIC, json.dumps(payload, ensure_ascii=False))
  print("Successful send message!")
  return True
if __name__ == '__main__':
  msg = "我是一条测试数据！"
  clicent_main(msg)
client

Python实现MQTT服务端代码

服务端


#!/usr/bin/env python 
# encoding: utf-8 
""" 
@version: v1.0 
@author: W_H_J 
@license: Apache Licence 
@contact: 415900617@qq.com 
@software: PyCharm 
@file: serverMqttTest.py 
@time: 2019/2/22 14:35 
@describe: mqtt 服务端
"""
import json
import sys
import os
import time
import paho.mqtt.client as mqtt
sys.path.append(os.path.abspath(os.path.dirname(__file__) + '/' + '..'))
sys.path.append("..")
 
REPORT_TOPIC = 'test' # 主题
 
 
def on_connect(client, userdata, flags, rc):
  print('connected to mqtt with resurt code ', rc)
  client.subscribe(REPORT_TOPIC) # 订阅主题
 
 
def on_message(client, userdata, msg):
  """
  接收客户端发送的消息
  :param client: 连接信息
  :param userdata: 
  :param msg: 客户端返回的消息
  :return: 
  """
  print("Start server!")
  payload = json.loads(msg.payload.decode('utf-8'))
  print(payload)
 
 
def server_conenet(client):
  client.on_connect = on_connect # 启用订阅模式
  client.on_message = on_message # 接收消息
  client.connect("127.0.0.1", 1883, 60) # 链接
  # client.loop_start()  # 以start方式运行，需要启动一个守护线程，让服务端运行，否则会随主线程死亡
  client.loop_forever()  # 以forever方式阻塞运行。
 
 
def server_stop(client):
  client.loop_stop() # 停止服务端
  sys.exit(0)
 
 
def server_main():
  client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
  client = mqtt.Client(client_id, transport='tcp')
  server_conenet(client)
 
 
if __name__ == '__main__':
  # 启动监听
  server_main()

server