tencent cloud

消息队列 MQTT 版

产品简介
TDMQ 产品系列介绍与选型
什么是消息队列 MQTT 版
应用场景
技术架构
产品系列
MQTT 协议兼容说明
开源对比
高可用
产品约束与使用配额
基本概念
开服地域
购买指南
计费概述
续费说明
查看消费明细
欠费说明
退费说明
快速入门
入门流程指引
准备工作
公网接入
VPC 网络接入
用户指南
使用流程指引
配置账号权限
新建集群
管理 Topic
连接集群
查询消息
管理客户端
管理集群
查看监控和配置告警
数据集成
集成数据到云函数 SCF
集成数据到 CKafka
集成数据到 RocketMQ
开发指南
MQTT 5 高级特性
数据面 HTTP 接口说明
配置自定义域名
配置 SQL 过滤
配置点对点订阅
MQTT over QUIC
管理客户端订阅
消息增强规则
实践教程
MQTT 客户端开发注意事项
可观测能力
Topic 与通配符订阅
API 参考
History
Introduction
API Category
Making API Requests
Cluster APIs
Topic APIs
Authorization Policy APIs
User APIs
Client APIs
Message Enhancement Rule APIs
Message APIs
Data Types
Error Codes
SDK 参考
接入点格式
Java SDK
C SDK
Javascript/Node.JS/小程序
Go SDK
iOS SDK
JavaScript SDK
Dart SDK
Python SDK
.NET
安全与合规
权限管理
常见问题
相关协议
隐私协议
数据处理和安全协议
消息队列 MQTT 版服务等级协议
联系我们

Python SDK

聚焦模式
字号
最后更新时间: 2026-01-30 15:23:30

功能概述

Eclipse Paho Python 为 Eclipse Paho 项目下的 Python 语言版客户端库,该库能够连接到 MQTT Broker 以发布消息,订阅主题并接收已发布的消息。
客户端依赖于 Google 的 proxywebsockets 软件包,通过以下命令完成安装:

云资源准备

请您先参见 创建资源 操作步骤完成云资源准备。

环境准备

mkdir -p /root/quick-start && python3 -m venv /root/quick-start
cd /root/quick-start
./bin/pip3 install paho-mqtt
注意:
paho-mqtt 仅支持 Python 3.7+。

示例代码

将下面代码保存到 /root/quick-start/example.py
# python 3.11

import time
import logging

from paho.mqtt import client as mqtt_client


host = 'mqtt-sample-sh-public.mqtt.tencenttdmq.com'
port = 1883
topic = "home/room/1"
client_id = 'QuickStart'
username = 'your-username'
password = 'your-password'

FIRST_RECONNECT_DELAY = 1
RECONNECT_RATE = 2
MAX_RECONNECT_COUNT = 12
MAX_RECONNECT_DELAY = 60

def on_disconnect(client, userdata, rc, properties):
logging.info("Disconnected with result code: %s", rc)
reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY
while reconnect_count < MAX_RECONNECT_COUNT:
logging.info("Reconnecting in %d seconds...", reconnect_delay)
time.sleep(reconnect_delay)

try:
client.reconnect()
logging.info("Reconnected successfully!")
return
except Exception as err:
logging.error("%s. Reconnect failed. Retrying...", err)

reconnect_delay *= RECONNECT_RATE
reconnect_delay = min(reconnect_delay, MAX_RECONNECT_DELAY)
reconnect_count += 1
logging.info("Reconnect failed after %s attempts. Exiting...", reconnect_count)

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

def connect_mqtt():
def on_connect(client, userdata, flags, rc, properties):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\\n", rc)
logging.basicConfig(level=logging.DEBUG)
client = mqtt_client.Client(mqtt_client.CallbackAPIVersion.VERSION2, client_id, clean_session=True, userdata=None, protocol=mqtt_client.MQTTv311)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message
client.enable_logger()
client.connect(host, port)
return client

def on_subscribe(client, userdata, mid, reason_code_list, properties):
# Since we subscribed only for a single channel, reason_code_list contains
# a single entry
if reason_code_list[0].is_failure:
print(f"Broker rejected you subscription: {reason_code_list[0]}")
else:
print(f"Broker granted the following QoS: {reason_code_list[0].value}")

def on_unsubscribe(client, userdata, mid, reason_code_list, properties):
# Be careful, the reason_code_list is only present in MQTTv5.
# In MQTTv3 it will always be empty
if len(reason_code_list) == 0 or not reason_code_list[0].is_failure:
print("unsubscribe succeeded (if SUBACK is received in MQTTv3 it success)")
else:
print(f"Broker replied with failure: {reason_code_list[0]}")

def subscribe(client):
client.on_subscribe = on_subscribe
client.on_unsubscribe = on_unsubscribe
# Subscribe topic with QoS 1
client.subscribe(topic, 1)
print(f"Subscribed `{topic}`")


def publish(client):
msg_count = 1
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1
if msg_count > 5:
break


def run():
client = connect_mqtt()
client.loop_start()
subscribe(client)
publish(client)
client.loop_stop()
time.sleep(30)


if __name__ == '__main__':
run()


运行示例

cd /root/quick-start
./bin/python3 example.py

样例输出

DEBUG:paho.mqtt.client:Sending CONNECT (u1, p1, wr0, wq0, wf0, c1, k60) client_id=b'quick-start'
DEBUG:paho.mqtt.client:Sending SUBSCRIBE (d0, m1) [(b'home/room/1', 1)]
Subscribed `home/room/1`
DEBUG:paho.mqtt.client:Received CONNACK (0, 0)
Connected to MQTT Broker!
DEBUG:paho.mqtt.client:Received SUBACK
Broker granted the following QoS: 1
DEBUG:paho.mqtt.client:Sending PUBLISH (d0, q0, r0, m2), 'b'home/room/1'', ... (11 bytes)
Send `messages: 1` to topic `home/room/1`
DEBUG:paho.mqtt.client:Received PUBLISH (d0, q0, r0, m0), 'home/room/1', ... (11 bytes)
Received `messages: 1` from `home/room/1` topic
DEBUG:paho.mqtt.client:Sending PUBLISH (d0, q0, r0, m3), 'b'home/room/1'', ... (11 bytes)
Send `messages: 2` to topic `home/room/1`
DEBUG:paho.mqtt.client:Received PUBLISH (d0, q0, r0, m0), 'home/room/1', ... (11 bytes)
Received `messages: 2` from `home/room/1` topic
DEBUG:paho.mqtt.client:Sending PUBLISH (d0, q0, r0, m4), 'b'home/room/1'', ... (11 bytes)
Send `messages: 3` to topic `home/room/1`
DEBUG:paho.mqtt.client:Received PUBLISH (d0, q0, r0, m0), 'home/room/1', ... (11 bytes)
Received `messages: 3` from `home/room/1` topic
DEBUG:paho.mqtt.client:Sending PUBLISH (d0, q0, r0, m5), 'b'home/room/1'', ... (11 bytes)
Send `messages: 4` to topic `home/room/1`
DEBUG:paho.mqtt.client:Received PUBLISH (d0, q0, r0, m0), 'home/room/1', ... (11 bytes)
Received `messages: 4` from `home/room/1` topic
DEBUG:paho.mqtt.client:Sending PUBLISH (d0, q0, r0, m6), 'b'home/room/1'', ... (11 bytes)
Send `messages: 5` to topic `home/room/1`
Received `messages: 5` from `home/room/1` topic


帮助和支持

本页内容是否解决了您的问题?

填写满意度调查问卷,共创更好文档体验。

文档反馈