class MQTTClient:
    def __init__(self, broker, port, username, password, ca_cert, topics):
        self.client = mqtt.Client()
        self.client.username_pw_set(self.username, self.password)
        self.client.tls_set(ca_certs=self.ca_cert)
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message

def set_message_handler(self, handler):
    self.custom_message_handler = handler

async def start_async(self):
    self.connect()
    await asyncio.get_event_loop().run_in_executor(None, self.client.loop_forever)

async def on_mqtt_message(topic, payload):
    print(f"Custom handler: {topic} -> {payload}")

mqtt_client.set_message_handler(on_mqtt_message)
await mqtt_client.start_async()

pip install paho-mqtt

python main.py

import paho.mqtt.client as mqtt
from datetime import datetime
import asyncio

class MQTTClient:
    def __init__(self, broker, port, username, password, ca_cert, topics):
        """
        初始化 MQTT 客戶端
        """
        self.broker = broker
        self.port = port
        self.username = username
        self.password = password
        self.ca_cert = ca_cert
        self.topics = topics
        self.client = mqtt.Client()

        # 配置 MQTT 客戶端
        self.client.username_pw_set(self.username, self.password)
        self.client.tls_set(ca_certs=self.ca_cert)
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message

        self.custom_message_handler = None  # 自定義消息處理器

    def set_message_handler(self, handler):
        """
        設(shè)置自定義消息處理回調(diào)函數(shù)
        """
        self.custom_message_handler = handler

    def on_connect(self, client, userdata, flags, rc):
        """
        連接成功時(shí)的回調(diào)
        """
        if rc == 0:
            print("SSL連接成功")
            for topic in self.topics:
                client.subscribe(topic)
                print(f"已訂閱主題: {topic}")
        else:
            print(f"連接失敗，返回碼: {rc}")

    def on_message(self, client, userdata, msg):
        """
        收到消息時(shí)的回調(diào)
        """
        current_time = datetime.now()
        payload = msg.payload.decode()
        print(f"收到消息: {msg.topic} -> {payload} 時(shí)間: {current_time}")

        if self.custom_message_handler and self.event_loop:
            asyncio.run_coroutine_threadsafe(
                self.custom_message_handler(msg.topic, payload),
                self.event_loop
            )

    def connect(self):
        """
        連接到 MQTT 服務(wù)器
        """
        self.client.connect(self.broker, self.port, keepalive=60)

    async def start_async(self):
        """
        異步運(yùn)行 MQTT 客戶端
        """
        self.connect()  # 確保連接到 MQTT 服務(wù)器
        print("Starting MQTT client loop...")

        # 異步運(yùn)行 MQTT 客戶端的事件循環(huán)
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, self.client.loop_forever)

import asyncio
from mqtt import MQTTClient

# MQTT 配置
MQTT_BROKER = "你的服務(wù)器地址"
MQTT_PORT = 8883  # 使用 SSL 的端口
MQTT_USERNAME = "用戶名"
MQTT_PASSWORD = "密碼"
CA_CERT = "./emqxsl-ca.crt"  # CA 證書(shū)路徑
TOPICS = ["clients/disconnect", "uhome/esp32"]  # 訂閱的主題列表


async def main():
    loop = asyncio.get_running_loop()

    mqtt_client = MQTTClient(
        broker=MQTT_BROKER,
        port=MQTT_PORT,
        username=MQTT_USERNAME,
        password=MQTT_PASSWORD,
        ca_cert=CA_CERT,
        topics=TOPICS
    )

    async def on_mqtt_message(topic, payload):
        print(f"Custom handler: {topic} -> {payload}")

    mqtt_client.set_message_handler(on_mqtt_message)
    mqtt_client.event_loop = loop  # 將事件循環(huán)傳遞給 MQTT 客戶端
    await mqtt_client.start_async()


    await asyncio.gather(websocket_task, periodic_task)


if __name__ == "__main__":
    asyncio.run(main())