Python 是一种广泛使用的解释型、高级编程、通用型编程语言。Python 的设计哲学强调代码的可读性和简洁的语法(尤其是使用空格缩进划分代码块,而非使用大括号或者关键词)。Python 让开发者能够用更少的代码表达想法,不管是小型还是大型程序,该语言都试图让程序的结构清晰明了。
MQTT 是一种基于发布/订阅模式的 轻量级物联网消息传输协议 ,可以用极少的代码和带宽为联网设备提供实时可靠的消息服务,它广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等行业。
本文主要介绍如何在 Python 项目中使用 paho-mqtt 客户端库 ,实现客户端与 MQTT 服务器的连接、订阅、取消订阅、收发消息等功能。
项目初始化
本项目使用 Python 3.6 进行开发测试,读者可用如下命令确认 Python 的版本。
"htmlcode">pip3 install -i https://pypi.doubanio.com/simple paho-mqttPython MQTT 使用
连接 MQTT 服务器
本文将使用 EMQ X 提供的 免费公共 MQTT 服务器 ,该服务基于 EMQ X 的 MQTT 物联网云平台 创建。服务器接入信息如下:
- Broker: broker.emqx.io
- TCP Port: 1883
- Websocket Port: 8083
导入 Paho MQTT客户端
from paho.mqtt import client as mqtt_client
设置 MQTT Broker 连接参数
设置 MQTT Broker 连接地址,端口以及 topic,同时我们调用 Python random.randint 函数随机生成 MQTT 客户端 id。
broker = 'broker.emqx.io' port = 1883 topic = "/python/mqtt" client_id = f'python-mqtt-{random.randint(0, 1000)}'
编写 MQTT 连接函数
编写连接回调函数 on_connect ,该函数将在客户端连接后被调用,在该函数中可以依据 rc 来判断客户端是否连接成功。通常同时我们将创建一个 MQTT 客户端,该客户端将连接到 broker.emqx.io 。
def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc == 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d\n", rc) # Set Connecting Client ID client = mqtt_client.Client(client_id) client.on_connect = on_connect client.connect(broker, port) return client
发布消息
首先定义一个 while 循环语句,在循环中我们将设置每秒调用 MQTT 客户端 publish 函数向 /python/mqtt 主题发送消息。
def publish(client): msg_count = 0 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
订阅消息
编写消息回调函数 on_message ,该函数将在客户端从 MQTT Broker 收到消息后被调用,在该函数中我们将打印出订阅的 topic 名称以及接收到的消息内容。
def subscribe(client: mqtt_client): def on_message(client, userdata, msg): print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic") client.subscribe(topic) client.on_message = on_message
完整代码
消息发布代码
# python 3.6 import random import time from paho.mqtt import client as mqtt_client broker = 'broker.emqx.io' port = 1883 topic = "/python/mqtt" # generate client ID with pub prefix randomly client_id = f'python-mqtt-{random.randint(0, 1000)}' def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc == 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d\n", rc) client = mqtt_client.Client(client_id) client.on_connect = on_connect client.connect(broker, port) return client def publish(client): msg_count = 0 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 def run(): client = connect_mqtt() client.loop_start() publish(client) if __name__ == '__main__': run()
消息订阅代码
# python 3.6 import random import time from paho.mqtt import client as mqtt_client broker = 'broker.emqx.io' port = 1883 topic = "/python/mqtt" # generate client ID with pub prefix randomly client_id = f'python-mqtt-{random.randint(0, 1000)}' def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc == 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d\n", rc) client = mqtt_client.Client(client_id) client.on_connect = on_connect client.connect(broker, port) return client def publish(client): msg_count = 0 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 def run(): client = connect_mqtt() client.loop_start() publish(client) if __name__ == '__main__': run() 消息订阅代码 # python3.6 import random from paho.mqtt import client as mqtt_client broker = 'broker.emqx.io' port = 1883 topic = "/python/mqtt" # generate client ID with pub prefix randomly client_id = f'python-mqtt-{random.randint(0, 100)}' def connect_mqtt() -> mqtt_client: def on_connect(client, userdata, flags, rc): if rc == 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d\n", rc) client = mqtt_client.Client(client_id) client.on_connect = on_connect client.connect(broker, port) return client def subscribe(client: mqtt_client): def on_message(client, userdata, msg): print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic") client.subscribe(topic) client.on_message = on_message def run(): client = connect_mqtt() subscribe(client) client.loop_forever() if __name__ == '__main__': run()
测试
消息发布
运行 MQTT 消息发布代码,我们将看到客户端连接成功,并且成功将消息发布。
python3 pub.py
消息订阅
运行 MQTT 消息订阅代码,我们将看到客户端连接成功,并且成功接收到发布的消息。
python3 sub.py
总结
至此,我们完成了使用 paho-mqtt 客户端连接到 公共 MQTT 服务器 ,并实现了测试客户端与 MQTT 服务器的连接、消息发布和订阅。
与 C ++ 或 Java 之类的高级语言不同,Python 比较适合设备侧的业务逻辑实现,使用 Python 您可以减少代码上的逻辑复杂度,降低与设备的交互成本。我们相信在物联网领域 Python 将会有更广泛的应用。
接下来我们将会陆续发布更多关于物联网开发及 Python 的相关文章,敬请关注。
以上就是在 Python 中使用 MQTT的方法的详细内容,更多关于Python 中使用 MQTT的资料请关注其它相关文章!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
稳了!魔兽国服回归的3条重磅消息!官宣时间再确认!
昨天有一位朋友在大神群里分享,自己亚服账号被封号之后居然弹出了国服的封号信息对话框。
这里面让他访问的是一个国服的战网网址,com.cn和后面的zh都非常明白地表明这就是国服战网。
而他在复制这个网址并且进行登录之后,确实是网易的网址,也就是我们熟悉的停服之后国服发布的暴雪游戏产品运营到期开放退款的说明。这是一件比较奇怪的事情,因为以前都没有出现这样的情况,现在突然提示跳转到国服战网的网址,是不是说明了简体中文客户端已经开始进行更新了呢?
更新日志
- 国风大师纯音系列《古筝仙子 常静演奏 十指弹一》1CD[FLAC][919MB]
- DasKapital-OneMustHaveChaosInsidetoGiveBirthtoaDancingStar(2024)[24-44.1]WAV
- 王菲.1999-《只爱陌生人》日本东芝版[低速原抓WAV+CUE]
- 张小英1987-金奖金曲第一集[新加坡版][WAV+CUE]
- 鹿晗.2015-重启【海蝶】【FLAC分轨】
- 张芸京.2012-小女孩【金牌大风】【WAV+CUE】
- 余天.2002-雄狮魅力金曲2CD(南方金点系列)【南方】【WAV+CUE】
- 银霞.1982-《结·SAYYESMYBOY》台湾复刻版[WAV+CUE]
- 梁汉文1993-火热动感[引进版][WAV+CUE]
- 梁汉文2008-EDMONDHITS48[香港]3CD[WAV+CUE]
- 国风大师纯音系列《童丽-[渭城曲]》1CD[WAV分轨][593.5MB]
- 国风大师纯音系列《许菱子《烧筝2》古筝曲集》1CD[WAV分轨][1.3GB]
- 王子异《1996 Part.2》[320K/MP3][18.29MB]
- 群星.1994-宝丽金最倾情3CD【宝丽金】【WAV+CUE】
- 乐俊亨.2024-同泪【CharioticMusic】【FLAC分轨】