物联网（IoT）正在改变全球的工作和生活方式。为了实现这种互联和通信，我们需要强大、灵活且可靠的工具。本文将探索如何结合Python的强大功能和MQTT的高效通信，为物联网项目提供无缝解决方案。

为何选择Python与MQTT？

Python，这一简洁、高效的编程语言，由于其出色的库支持和跨平台特性，已经成为了数据科学、网络开发和自动化的首选。它简单的语法和代码可读性使得开发过程更为迅速和直观。

另一方面，MQTT是一种轻量级的发布/订阅模式的通信协议，特别适用于低带宽、高延迟或不稳定的网络环境。正因为其高效和低耗，MQTT已成为物联网通信的标准。

步骤详解：构建基于Python的MQTT物联网应用

1. 环境准备

确保已安装Python 3.6或更高版本。你可以使用以下命令确认：

python3 --version
2. 安装MQTT库

我们将使用paho-mqtt，这是一个流行的Python MQTT客户端库。

pip3 install -i https://pypi.doubanio.com/simple paho-mqtt
3. 连接到MQTT服务器

在此，我们选择使用EMQX提供的免费MQTT公共服务器，但同样可以选择其他任何MQTT broker。

  • 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()

消息订阅代码

# 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
如何用Python实现MQTT通讯-MQTT中文站

消息订阅

运行 MQTT 消息订阅代码，我们将看到客户端连接成功，并且成功接收到发布的消息。

python3 sub.py
如何用Python实现MQTT通讯-MQTT中文站

扩展你的IoT应用

  • 数据安全：使用SSL/TLS保证数据的安全传输。
  • 数据存储：结合数据库技术，例如SQLite、MySQL或MongoDB，持久化存储设备数据。
  • 实时分析：利用Python的强大数据分析库，如pandas、NumPy和matplotlib，对接收到的数据进行实时处理和可视化。
  • 设备管理：考虑添加一个设备管理系统，以便跟踪、更新和维护所有连接的设备。

总结

物联网的未来取决于我们如何有效地处理海量的设备数据。Python和MQTT为我们提供了一个高效、简洁且可靠的方式来捕获、分析和共享这些数据。无论是智能家居、工业自动化还是智慧城市，结合Python和MQTT的IoT解决方案都将开创新的可能性和机会。

打赏微海报分享