137 lines
4.9 KiB
Python
137 lines
4.9 KiB
Python
import paho.mqtt.client as mqtt
|
||
import uuid
|
||
from config.logger_config import logger
|
||
from config.setting import settings
|
||
|
||
|
||
class MQTTClient:
|
||
def __init__(self):
|
||
self.broker_host = settings.mqtt_host
|
||
self.broker_port = settings.mqtt_port
|
||
self.client_id = self.generate_client_id()
|
||
|
||
self.client = mqtt.Client(client_id=self.client_id)
|
||
|
||
# 注册回调函数
|
||
self.client.on_connect = self._on_connect
|
||
self.client.on_disconnect = self._on_disconnect
|
||
self.client.on_message = self._on_message
|
||
|
||
# 存储消息处理器的字典，键为主题，值为处理函数
|
||
self.message_handlers = {}
|
||
|
||
self.connected = False
|
||
|
||
def _on_connect(self, client, userdata, flags, rc):
|
||
"""连接回调函数"""
|
||
self.connected = rc == 0
|
||
if self.connected:
|
||
logger.info(f"成功连接到MQTT broker {self.broker_host}:{self.broker_port}")
|
||
else:
|
||
logger.error(f"连接MQTT broker失败，错误代码: {rc}")
|
||
|
||
def _on_disconnect(self, client, userdata, rc):
|
||
"""断开连接回调函数"""
|
||
self.connected = False
|
||
if rc != 0:
|
||
logger.warning(f"意外断开与MQTT broker的连接，错误代码: {rc}")
|
||
else:
|
||
logger.info("已与MQTT broker断开连接")
|
||
|
||
def _on_message(self, client, userdata, msg):
|
||
"""消息接收回调函数"""
|
||
logger.info(f"收到消息: {msg.topic}")
|
||
try:
|
||
if msg.topic in self.message_handlers:
|
||
self.message_handlers[msg.topic](client, msg)
|
||
else:
|
||
logger.warning(f"未找到 {msg.topic} 的消息处理器")
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理消息时出错: {str(e)}", exc_info=True)
|
||
|
||
@staticmethod
|
||
def generate_client_id(prefix: str = "client") -> str:
|
||
"""生成唯一客户端ID"""
|
||
uuid_str = str(uuid.uuid4()).split('-')[0]
|
||
return f"{prefix}-{uuid_str}"
|
||
|
||
def add_message_handler(self, topic, handler):
|
||
"""添加消息处理器"""
|
||
if topic in self.message_handlers:
|
||
logger.warning(f"主题 {topic} 已存在处理器，将被覆盖")
|
||
self.message_handlers[topic] = handler
|
||
logger.info(f"为主题 {topic} 注册了消息处理器")
|
||
logger.debug(f"当前消息处理器: {self.message_handlers.keys()}")
|
||
|
||
def connect(self):
|
||
"""连接到MQTT服务器"""
|
||
try:
|
||
self.client.connect(self.broker_host, self.broker_port)
|
||
self.start_loop()
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"连接MQTT broker {self.broker_host}:{self.broker_port}时发生错误: {str(e)}", exc_info=True)
|
||
return False
|
||
|
||
def disconnect(self):
|
||
"""断开与MQTT服务器的连接"""
|
||
try:
|
||
self.stop_loop()
|
||
self.client.disconnect()
|
||
logger.info("正在断开与MQTT broker的连接")
|
||
except Exception as e:
|
||
logger.error(f"断开连接时发生错误: {str(e)}", exc_info=True)
|
||
|
||
def subscribe(self, topic, qos=0):
|
||
"""订阅主题"""
|
||
try:
|
||
result, mid = self.client.subscribe(topic, qos)
|
||
if result == mqtt.MQTT_ERR_SUCCESS:
|
||
logger.info(f"已订阅主题: {topic} (QoS: {qos})")
|
||
return True
|
||
else:
|
||
logger.error(f"订阅主题 {topic} 失败，错误代码: {result}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"订阅主题时发生错误: {str(e)}", exc_info=True)
|
||
return False
|
||
|
||
def publish(self, topic, payload, qos=0, retain=False):
|
||
"""发布消息到指定主题"""
|
||
if not self.connected:
|
||
logger.warning("未连接到MQTT broker，无法发布消息")
|
||
return False
|
||
|
||
try:
|
||
result = self.client.publish(topic, payload, qos, retain)
|
||
result.wait_for_publish()
|
||
if result.rc == mqtt.MQTT_ERR_SUCCESS:
|
||
logger.debug(f"已发布消息到主题 {topic}: {payload}")
|
||
return True
|
||
else:
|
||
logger.error(f"发布消息到主题 {topic} 失败，错误代码: {result.rc}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"发布消息时发生错误: {str(e)}", exc_info=True)
|
||
return False
|
||
|
||
def start_loop(self):
|
||
"""启动MQTT网络循环"""
|
||
try:
|
||
self.client.loop_start()
|
||
logger.info("已启动MQTT网络循环")
|
||
except Exception as e:
|
||
logger.error(f"启动网络循环时发生错误: {str(e)}", exc_info=True)
|
||
|
||
def stop_loop(self):
|
||
"""停止MQTT网络循环"""
|
||
try:
|
||
self.client.loop_stop()
|
||
logger.info("已停止MQTT网络循环")
|
||
except Exception as e:
|
||
logger.error(f"停止网络循环时发生错误: {str(e)}", exc_info=True)
|
||
|
||
|
||
mqtt_client = MQTTClient()
|