欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Python實現(xiàn)MQTT通信的示例代碼

 更新時間：2025年07月18日 11:13:33   作者：weixin_43182240  
本文主要介紹了Python實現(xiàn)MQTT通信的示例代碼,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

 適用于物聯(lián)網(wǎng)設備、傳感器數(shù)據(jù)采集等場景

1. 安裝paho-mqtt庫‌

安裝Python的MQTT客戶端庫 paho-mqtt，支持MQTT v3.1/v3.1.1協(xié)議，兼容性強

pip install paho-mqtt==1.6.1  # 推薦使用1.6.1版本避免兼容性問題

2. 搭建MQTT代理服務器（Broker）‌

選擇本地搭建或使用公共Broker：

‌本地搭建（以EMQX為例）‌

下載并解壓EMQX開源版。Directory listing for EMQX: /v5.3.2/ | EMQ

配置EMQX，EMQX的配置文件位于etc\emqx.conf，可以根據(jù)需要修改配置文件，例如更改端口號、添加插件等：18083 EMQX Dashboard 管理控制臺端口

在命令行中進入安裝目錄的 bin 文件夾，執(zhí)行以下命令啟動服務

檢查啟動狀態(tài)emqx_ctl status

訪問 http://localhost:18083 進入管理界面（默認賬號：admin/public）

公共Broker（如HiveMQ）‌

使用免費公共服務器： 

broker_address = "broker.hivemq.com"  
broker_port = 1883

‌3. Python實現(xiàn)基礎通信功能‌

分為 ‌發(fā)布者（Publisher）‌ 和 ‌訂閱者（Subscriber）‌ 兩類客戶端：

‌3.1 發(fā)布者代碼示例publisher.py

# python 3.8

import random
import time

from paho.mqtt import client as mqtt_client


broker = 'broker.emqx.io'
port = 1883
broker = '127.0.0.1'
port = 1883
topic = "python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
username = 'emqx'
password = '**********'


def connect_mqtt():
    def on_connect(client, userdata, flags, rc,properties):
        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 = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)
    # client.tls_set(ca_certs='./server-ca.crt')
    # client.username_pw_set(username, password)
    client.username_pw_set('admin', 'public')
    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()

3.2 訂閱者代碼示例subscriber.py

# python3.8

import random

from paho.mqtt import client as mqtt_client


broker = 'broker.emqx.io'
port = 1883
broker = '127.0.0.1'
port = 1883
topic = "python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 100)}'
username = 'emqx'
password = '**********'


def connect_mqtt() -> mqtt_client:
    def on_connect(client, userdata, flags, rc,properties):
        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 = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)
    # client.tls_set(ca_certs='./server-ca.crt')
    # client.username_pw_set(username, password)
    client.username_pw_set('admin', 'public')
    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()

3.3或者放同一個文件：

from paho.mqtt import client as mqtt_client
from threading import Thread
import time

broker = '127.0.0.1'
port = 1883
# broker = 'broker.emqx.io'
# port = 1883
su_topic = [("python_mqtt_server", 2), ("python_mqtt_command", 2)]  # 訂閱頻道,數(shù)組格式為(topic，qos)
pu_topic = ["python_mqtt_server"]  # 發(fā)布頻道
client_id = 'websocket_publisher_001'


# 將連接與訂閱綁在一起，防止復用client導致連接沖突
def connect_mqtt():
    # For paho-mqtt 2.0.0, you need to add the properties parameter.
    def on_connect(client, userdata, flags, rc, properties):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    # For paho-mqtt 2.0.0, you need to set callback_api_version.
    client = mqtt_client.Client(client_id=client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)

    client.username_pw_set('admin', 'public')
    client.on_connect = on_connect
    client.connect(broker, port)
    subscribe_many(client)
    print("訂閱完成")
    return client


# 訂閱多個頻道
def subscribe_many(client):
    # 根據(jù)client源碼可知，訂閱支持的數(shù)組格式是(topic,qos)
    client.subscribe(su_topic)


# 收到信息的回調(diào)函數(shù)
def on_message(client, userdata, msg):
    print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
    # 這里可以編寫一些收到信息后的處理,比如收到什么信息開啟什么任務，任務完成后再調(diào)publish發(fā)布信息給終端等等........
    # ----------處理例子------------------
    r_msg = msg.payload.decode()
    if int(r_msg) % 2 == 0:
        print("收到偶數(shù)")
    # -----------------------------------


# 發(fā)布信息函數(shù)
def publish(client):
    for i in range(1, 11):
        time.sleep(1)
        msg = f"{i}"
        result = client.publish(pu_topic[0], msg)
        # result: [0, 1]
        status = result[0]
        if status == 0:
            print(f"Send `{msg}` to topic `{pu_topic[0]}`")
        else:
            print(f"Failed to send message to topic {pu_topic[0]}")


if __name__ == '__main__':
    client = connect_mqtt()
    client.username_pw_set('admin','public')
    client.on_message = on_message  # 重寫回調(diào)函數(shù),可以在方法體里，根據(jù)需求改收到信息后干什么
    # 多線程，這里啟動一個發(fā)布任務的線程，模擬有一個發(fā)布消息的客戶端
    thread1 = Thread(target=publish, args=(client,))
    # --*如要要實現(xiàn)流水線式的收發(fā)，則可以將線程內(nèi)方法改為任務隊列處理，用一個全局變量管理任務執(zhí)行，執(zhí)行后再publish
    thread1.start()

    client.loop_forever()  # 這個要加,監(jiān)聽(on_message)與loop_forever配套

關鍵配置說明

訂閱所有傳感器子主題

client.subscribe("sensors/#", qos=1) ?

‌WebSocket協(xié)議支持‌

  • 必須設置 transport="websockets"，否則默認使用TCP連接。
  • 端口需與Broker配置匹配（例如EMQX默認WebSocket端口為8083）。

‌QoS與消息保留‌

  • qos=1：確保消息至少送達一次。
  • retain=True：Broker保留最后一條消息，新訂閱者立即收到。

‌異常處理‌

  • 使用 try-except 捕獲網(wǎng)絡中斷或Broker不可用問題。
  • loop_start() 適用于需要異步處理的場景（如GUI應用）。

4. 關鍵功能擴展

異步通信，使用多線程實現(xiàn)非阻塞操作：

from threading import Thread  

  def start_async(client):  
      Thread(target=client.loop_start).start()

服務質(zhì)量（QoS）設置 支持至多一次（0）、至少一次（1）、恰好一次（2）三種級別：

client.publish(topic, payload, qos=1)  # 設置QoS為1

5. 注意事項‌

  1. ‌連接安全性‌：若需SSL加密，可使用 client.tls_set() 方法。
  2. ‌主題設計‌：遵循分層結構（如 sensor/room1/temp），支持通配符 +（單層）和 #（多層）。
  3. ‌異常處理‌：添加 on_disconnect 回調(diào)函數(shù)實現(xiàn)斷線重連邏輯。

‌完整示例流程‌

  1. 啟動EMQX Broker服務。
  2. 運行訂閱者代碼監(jiān)聽主題 home/sensor/temp。
  3. 運行發(fā)布者代碼發(fā)送溫度數(shù)據(jù)。
  4. 在EMQX管理界面驗證消息收發(fā)狀態(tài)

到此這篇關于Python實現(xiàn)MQTT通信的示例代碼的文章就介紹到這了,更多相關Python MQTT通信內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！ 

相關文章

最新評論