"htmlcode">

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 的相关文章，敬请关注。
以上就是在 Python 中使用 MQTT的方法的详细内容，更多关于Python 中使用 MQTT的资料请关注其它相关文章！