使用paho.mqtt实现MQTT协议通信的最佳实践

发布时间：2023-05-21

MQTT（Message Queue Telemetry Transport）是一种基于发布/订阅模式的轻量级物联网通信协议，它在移动设备和物联网应用方面得到了广泛的应用。本文将介绍使用paho.mqtt实现MQTT协议通信的最佳实践，包括MQTT协议基础概念、MQTT客户端的连接建立、消息订阅和发布、QoS质量等级、TLS/SSL安全连接等。

一、MQTT协议基础概念

MQTT协议是一种基于TCP/IP协议的轻量级协议，具有以下特点：

  1. 发布/订阅模式：MQTT协议采用发布-订阅（publish-subscribe）模式，允许多个客户端监听同一个主题（Topic），在主题发送消息时所有订阅该主题的客户端都会接收到消息。
  2. 压缩消息：MQTT协议支持消息压缩，可以有效地减少网络传输的数据量。
  3. QoS质量等级：MQTT协议支持三种服务质量等级，分别为At most once、At least once和Exactly once。
  4. 低耗能：MQTT协议采用轻量级的二进制通信协议，协议开销小，适合在带宽较低、 CPU和内存资源有限的设备间通信。

二、MQTT客户端的连接建立

MQTT客户端的连接建立需要指定以下三个参数：

  1. Broker地址：MQTT Broker是实现MQTT协议的服务端，用于处理客户端的连接、认证和消息路由等。客户端需要指定Broker的地址。
  2. Client ID：客户端ID是组成客户端标识的字符串，客户端需要在连接Broker时指定Client ID，如果不指定，Broker会随机生成一个唯一的Client ID。
  3. Will Message：客户端可以指定一个遗嘱消息，当客户端异常断开连接时，Broker会将遗嘱消息发布到指定的主题，通知其他客户端。
import paho.mqtt.client as mqtt
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
WILL_MESSAGE = {"topic": "WillTopic", "payload": "Connection closed unexpectedly!", "qos": 2, "retain": False}
client = mqtt.Client(client_id=CLIENT_ID)
client.will_set(topic=WILL_MESSAGE["topic"], payload=WILL_MESSAGE["payload"], qos=WILL_MESSAGE["qos"], retain=WILL_MESSAGE["retain"])
client.connect(BROKER_ADDRESS)

三、消息订阅和发布

MQTT协议支持消息订阅和发布两种操作，客户端可以订阅指定的主题，也可以发布消息到指定的主题。

  1. 消息订阅：客户端订阅一个主题后，当该主题有消息发布时，该消息就会被该客户端接收到。
import paho.mqtt.client as mqtt
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
TOPIC = "test/topic"
QOS = 2
def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    client.subscribe(TOPIC, QOS)
def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))
client = mqtt.Client(client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER_ADDRESS)
client.loop_forever()
  1. 消息发布：客户端可以向指定主题发布消息，其他订阅了该主题的客户端会接收到该消息。
import paho.mqtt.client as mqtt
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
TOPIC = "test/topic"
PAYLOAD = "Hello, MQTT!"
QOS = 2
def on_publish(client, userdata, mid):
    print("Message published.")
client = mqtt.Client(client_id=CLIENT_ID)
client.on_publish = on_publish
client.connect(BROKER_ADDRESS)
client.publish(TOPIC, PAYLOAD, QOS)
client.loop_forever()

四、QoS质量等级

MQTT协议支持三种服务质量等级，分别为At most once、At least once和Exactly once。

  1. At most once：消息可能会丢失或重复发送，最多发送一次。
import paho.mqtt.client as mqtt
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
TOPIC = "test/topic"
PAYLOAD = "Hello, MQTT!"
QOS = 0
client = mqtt.Client(client_id=CLIENT_ID)
client.connect(BROKER_ADDRESS)
client.publish(TOPIC, PAYLOAD, QOS)
client.loop_forever()
  1. At least once：消息至少会被处理一次，但可能会重复发送。
import paho.mqtt.client as mqtt
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
TOPIC = "test/topic"
PAYLOAD = "Hello, MQTT!"
QOS = 1
def on_publish(client, userdata, mid):
    print("Message published.")
client = mqtt.Client(client_id=CLIENT_ID)
client.on_publish = on_publish
client.connect(BROKER_ADDRESS)
client.publish(TOPIC, PAYLOAD, QOS)
client.loop_forever()
  1. Exactly once：消息确保仅被处理一次。
import paho.mqtt.client as mqtt
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
TOPIC = "test/topic"
PAYLOAD = "Hello, MQTT!"
QOS = 2
def on_publish(client, userdata, mid):
    print("Message published.")
client = mqtt.Client(client_id=CLIENT_ID)
client.on_publish = on_publish
client.connect(BROKER_ADDRESS)
client.publish(TOPIC, PAYLOAD, QOS)
client.loop_forever()

五、TLS/SSL安全连接

MQTT协议支持使用TLS/SSL协议进行安全连接，提供端到端的数据加密和身份认证。

import paho.mqtt.client as mqtt
import ssl
BROKER_ADDRESS = "broker.emqx.io"
CLIENT_ID = "paho-mqtt-client"
TOPIC = "test/topic"
PAYLOAD = "Hello, MQTT!"
QOS = 2
def on_publish(client, userdata, mid):
    print("Message published.")
client = mqtt.Client(client_id=CLIENT_ID)
client.on_publish = on_publish
client.tls_set(ca_certs="ca.pem", certfile="client.crt", keyfile="client.key",
               cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None)
client.connect(BROKER_ADDRESS, port=8883)
client.publish(TOPIC, PAYLOAD, QOS)
client.loop_forever()

以上是使用paho.mqtt实现MQTT协议通信的最佳实践，在实际应用中可以根据实际需求进行调整和优化。

推荐文章