导航：首页 > 互联网科技 >

在树莓派上如何使用MQTT

发表于：2025-02-02 作者：千家信息网编辑

千家信息网最后更新 2025年02月02日，这篇文章主要为大家展示了"在树莓派上如何使用MQTT "，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下"在树莓派上如何使用MQTT "这篇文章吧。环境搭建安装

千家信息网最后更新 2025年02月02日在树莓派上如何使用MQTT

这篇文章主要为大家展示了"在树莓派上如何使用MQTT "，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下"在树莓派上如何使用MQTT "这篇文章吧。

环境搭建

安装 Python3

本项目使用 Python3 进行开发，一般情况下，树莓派系统会内置 Python3，如果不确定系统内是否已经安装，可以使用下面的命令进行确认。

python3 --version

如果显示 Python 3.x.x（x 表示数字）则表示已经安装，否则请使用 apt 命令安装（或跟随 Python3 安装指南操作）。

sudo apt install python3

安装 MQTT 客户端库

为了方便连接到 MQTT 服务器，我们需要安装 paho-mqtt 库。可以选择以下两种方法之一进行安装。

使用源码安装

git clone https://github.com/eclipse/paho.mqtt.python cd paho.mqtt.python python3 setup.py install

使用 pip3 安装

pip3 install paho-mqtt

MQTT 的使用

连接 MQTT 服务器

本文将使用 EMQ X 提供的免费公共 MQTT 服务器，该服务基于 EMQ X 的 MQTT 物联网云平台创建。服务器接入信息如下：

Broker: broker.emqx.io
TCP Port: 1883
Websocket Port: 8083

如果有需要，您也可以使用 docker 在本地快速安装 EMQ X 服务器。

docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 18083:18083 emqx/emqx

连接示例代码

# test_connect.py import paho.mqtt.client as mqtt # 回调函数。当尝试与 MQTT broker 建立连接时，触发该函数。# client 是本次连接的客户端实例。# userdata 是用户的信息，一般为空。但如果有需要，也可以通过 user_data_set 函数设置。# flags 保存服务器响应标志的字典。# rc 是响应码。# 一般情况下，我们只需要关注 rc 响应码是否为 0 就可以了。def on_connect(client, userdata, flags, rc):    if rc == 0:        print("Connected success")    else:        print(f"Connected fail with code {rc}")client = mqtt.Client() client.on_connect = on_connect client.connect("broker.emqx.io", 1883, 60) client.loop_forever()

将上面的代码保存为 test_connect.py 文件，并运行

python3 test_connect.py

我们在 on_connect 函数里对响应码进行了判断，为 0 则输出 Connected success 表示连接成功。如果返回的是其它数字，我们就需要对照下面的响应码进行判断。

0: 连接成功1: 连接失败-不正确的协议版本2: 连接失败-无效的客户端标识符3: 连接失败-服务器不可用4: 连接失败-错误的用户名或密码5: 连接失败-未授权6-255: 未定义如果是其它问题，可以检查网络情况，或者确认是否安装了 `paho-mqtt`。

在 MQTT 协议的概念中，消息是通过主题传递的，比如设备 A 向主题 T 发送消息，那么只有订阅了主题 T 的设备才能接收到。所以仅仅接入 MQTT 服务器并没有太大意议，要完整地使用 MQTT 服务，我们还需要知道如何订阅和发布消息。

订阅消息

打开任意编辑器，输入下面的代码，并保存为 subscriber.py 文件：

# subscriber.pyimport paho.mqtt.client as mqttdef on_connect(client, userdata, flags, rc):    print(f"Connected with result code {rc}")    # 订阅，需要放在 on_connect 里    # 如果与 broker 失去连接后重连，仍然会继续订阅 raspberry/topic 主题    client.subscribe("raspberry/topic")# 回调函数，当收到消息时，触发该函数def on_message(client, userdata, msg):    print(f"{msg.topic} {msg.payload}")    client = mqtt.Client()client.on_connect = on_connectclient.on_message = on_message# 设置遗嘱消息，当树莓派断电，或者网络出现异常中断时，发送遗嘱消息给其他客户端client.will_set('raspberry/status', {"status": "Off"})# 创建连接，三个参数分别为 broker 地址，broker 端口号，保活时间client.connect("broker.emqx.io", 1883, 60)# 设置网络循环堵塞，在调用 disconnect() 或程序崩溃前，不会主动结束程序client.loop_forever()

调用 subscribe() 函数，可以让树莓派订阅一个主题。在上面的代码中，我们使用它订阅了 raspberry/topic 主题，并监听消息。

另外，我们还使用 will_set() 设置了遗嘱消息。遗嘱消息是 MQTT 的一个特性，当设备在意外断开网络连接后，会向某个特定的主题发送消息。通过这个特性，我们可以得知树莓派是否断电，或者出现网络异常。

发布消息

打开任意编辑器，输入下面的代码，并保存为 publisher.py 文件：

import paho.mqtt.client as mqttimport timedef on_connect(client, userdata, flags, rc):    print(f"Connected with result code {rc}")    client = mqtt.Client()client.on_connect = on_connectclient.connect("broker.emqx.io", 1883, 60)# 每间隔 1 秒钟向 raspberry/topic 发送一个消息，连续发送 5 次for i in range(5):      # 四个参数分别为：主题，发送内容，QoS, 是否保留消息    client.publish('raspberry/topic', payload=i, qos=0, retain=False)    print(f"send {i} to raspberry/topic")    time.sleep(1)client.loop_forever()

调用 publish() 函数，可以向一个主题发送消息。在上面的代码中，我们使用了它向主题 raspberry/topic 发送消息。其中参数 QoS 是另一个 MQTT 特性，如果你想了解更多 QoS 的内容，可以查看 MQTT QoS（服务质量）介绍，这里我们暂且设为 0。