在 Python 中使用 MQTT的方法
作者:EMQX 发布时间:2022-01-26 12:32:48
目录
项目初始化
选择 MQTT 客户端库
Pip 安装 Paho MQTT 客户端
Python MQTT 使用
连接 MQTT 服务器
导入 Paho MQTT客户端
设置 MQTT Broker 连接参数
编写 MQTT 连接函数
发布消息
订阅消息
完整代码
消息发布代码
消息订阅代码
测试
消息发布
消息订阅
总结
Python 是一种广泛使用的解释型、高级编程、通用型编程语言。Python 的设计哲学强调代码的可读性和简洁的语法(尤其是使用空格缩进划分代码块,而非使用大括号或者关键词)。Python 让开发者能够用更少的代码表达想法,不管是小型还是大型程序,该语言都试图让程序的结构清晰明了。
MQTT 是一种基于发布/订阅模式的 轻量级物联网消息传输协议 ,可以用极少的代码和带宽为联网设备提供实时可靠的消息服务,它广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等行业。
本文主要介绍如何在 Python 项目中使用 paho-mqtt 客户端库 ,实现客户端与 MQTT 服务器的连接、订阅、取消订阅、收发消息等功能。
项目初始化
本项目使用 Python 3.6 进行开发测试,读者可用如下命令确认 Python 的版本。
➜ ~ python3 --version
Python 3.6.7
选择 MQTT 客户端库
paho-mqtt 是目前 Python 中使用较多的 MQTT 客户端库,它在 Python 2.7 或 3.x 上为客户端类提供了对 MQTT v3.1 和 v3.1.1 的支持。它还提供了一些帮助程序功能,使将消息发布到 MQTT 服务器变得非常简单。
Pip 安装 Paho MQTT 客户端
Pip 是 Python 包管理工具,该工具提供了对 Python 包的查找、下载、安装、卸载的功能。
pip3 install -i https://pypi.doubanio.com/simple paho-mqtt
Python 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 的相关文章,敬请关注。
来源:https://www.tuicool.com/articles/N3QvQbm
猜你喜欢
- 本文实例讲述了Python常见字符串操作函数。分享给大家供大家参考,具体如下:str.split(' ')1.按某一个字符分
- 可在图片连接后面加上宽和高,如:width="150" height="50"<h
- 本文实例为大家分享了python保存网页图片到本地的具体代码,供大家参考,具体内容如下#!/usr/bin/env Python#codin
- 插件说明:插件根据提供的10位ISBN书号,在Amazon网站上查找该图书的详细信息。如果找到结果,则返回一个两元素的数组,其中第一个元素是
- 在今天网络开发方面,JavaScript起了很关键的作用;像jQuery, MooTools, Prototype等等JavaScript框
- 本文实例讲述了Django框架视图介绍与使用。分享给大家供大家参考,具体如下:视图视图:即一个python函数,可以叫 视图函数,或者简称
- 随滚动条移动的DIV层js代码,无论你的滚动条到哪里这个DIV层就跟到哪里!代码中例举了五个方向的滚动div层例子:包括左上方的div,左下
- 一、元素偏移量 offset 系列offset 翻译过来就是偏移量, 我们使用 offset 系列相关属性可以动态的得到该元素的位置(偏移)
- SpringBoot体系内推荐使用Thymeleaf作为前端页面模板。jsp还得自己整合一下。1.项目结构对比以前的项目结构,main目录下
- 使用pip安装Django时报错,先是:C:\Users\admin>pip install django Collecting dj
- 前言在laravel项目开发中,经常使用到公共函数,那如何在laravel配置全局公共函数呢??下面话不多说了,来一起看看详细的介绍吧方法如
- Filed under 数据库技术Leave a commentSQL Server命令行导数据两种方式bcp和sqlcmd先说一下bcp:
- Jaslabs的Justin Silverton列出了十条有关优化MySQL查询的语句,我不得不对此发表言论,因为这个清单非常非常糟糕。另外
- 众所周知,Python 支持多种编程范式:过程式(使用基础的语句)、面向对象编程和函数式编程。Python 也提供了其他函数式编程语言的工具
- 本文实例讲述了python通过线程实现定时器timer的方法。分享给大家供大家参考。具体分析如下:这个python类实现了一个定时器效果,调
- OCR与Tesseract介绍将图片翻译成文字一般被称为光学文字识别(Optical Character Recognition,OCR)。
- 网上有许关于固定表格的标题行的文章,但是既要固定标题行又要固定标题列的却几乎没有。现我写下如下代码以供大家参考:<html> &
- 本文实例为大家分享了python实现网上购物系统的具体代码,供大家参考,具体内容如下1.购物商城的需求分析:1、输出欢迎界面还有登录注册菜单
- Python“json.decoder.JSONDecodeError: Expecting value: line 1
- 在进行爬虫爬取淘宝商品信息时候,利用selenium来模拟浏览器进行爬取时遇到了这个问题:selenium.common.exception