import threading import time import string from threading import Thread import menu_utils as utils import paho.mqtt.client as mqtt from print_color import * #mqtt 消息处理基类, mqtt消息处理 class class_comm_mqtt_interface : def __init__(self, name = "mqtt handler"): self.name = name def on_message(self, mqtt_thread, topic, message) : return def on_connect(self, mqtt_thread, userdata, flags, rc) : return #mqtt服务器连接处理 def mqtt_on_connect(client, userdata, flags, rc): if rc == 0: # 连接成功 print_normal_msg("Mqtt 服务器连接成功") elif rc == 1: # 协议版本错误 print_error_msg("协议版本错误") elif rc == 2: # 无效的客户端标识 print_error_msg("无效的客户端标识") elif rc == 3: # 服务器无法使用 print_error_msg("Mqtt 服务器无法使用") elif rc == 4: # 错误的用户名或密码 print_error_msg("错误的用户名或密码") elif rc == 5: # 未经授权 print_error_msg("未经授权") #获取线程类 （class_comm_mqtt_thread) mqtt_thread : class_comm_mqtt_thread= userdata unique_object : class_comm_mqtt_interface = None #遍历所有unique_object, 通知其 on_connect 事件 for unique_object in mqtt_thread.unique_object_dict.values() : if isinstance(unique_object, list) : #unique_object 可能是列表, 所以需要遍历 for each_object in unique_object: if hasattr(each_object, "on_connect") : each_object.on_connect(mqtt_thread, userdata, flags, rc) elif hasattr(unique_object, "on_connect") : unique_object.on_connect(mqtt_thread, userdata, flags, rc) #mqtt接收到主题, 通过主题中包含的 unique_name, 定位到 uniuqe_object, 并调用 unique_object.on_message 函数 def mqtt_on_message(client, userdata, message): #print("mqtt message:", message.topic, message.payload.decode("utf-8")) #获取线程类 （class_comm_mqtt_thread) mqtt_thread : class_comm_mqtt_thread= userdata unique_object : class_comm_mqtt_interface = None for unique_name, search_unique_object in mqtt_thread.unique_object_dict.items() : if unique_name in message.topic : unique_object = search_unique_object break #mqtt线程本身无法处理对应的主题消息, 由mqtt 所对应的unique_object处理 if unique_object != None : if isinstance(unique_object, list) : #unique_object 可能是列表, 所以需要遍历 for each_object in unique_object: if hasattr(each_object, "on_message") : each_object.on_message(mqtt_thread, message.topic, message.payload) elif hasattr(unique_object, "on_message") : unique_object.on_message(mqtt_thread, message.topic, message.payload) class class_comm_mqtt_thread(Thread): def __init__(self, user_name = "admin", password = "admin") : Thread.__init__(self) self.client = None self.subscribe_list = [] self.publish_list = [] self.publish_list_lock = threading.Lock() self.condition = threading.Condition() self.stop_request = False self.server = "127.0.0.1" self.port = 1883 self.user_name = user_name self.password = password self.keepalive = 60 self.unique_object_dict = {} #key = unique_name, value = device_object return #unique_object: 为unique_object_subscribe_name对应的object, 在回调 def add_unique_object(self, unique_name : string, unique_object : class_comm_mqtt_interface) : self.unique_object_dict[unique_name] = unique_object #topic 中 需要包含 def subscribe(self, topic : string) : if self.is_connect() and self.client: if topic not in self.subscribe_list : self.subscribe_list.append(topic) self.client.subscribe(topic) def set_mqtt_server(self, server = "127.0.0.1", port = 1883, keep_alive = 60.0, user_name = "admin", password = "admin") : self.server = server self.port = port self.user_name = user_name self.password = password self.keepalive = keep_alive #用于判断 mqtt_thread是否已经连接 def is_connect(self) : if self.client == None : return False return self.client.is_connected() def open(self): try : self.client = mqtt.Client() self.client.user_data_set(self) self.client.on_connect = mqtt_on_connect self.client.on_message = mqtt_on_message self.client.username_pw_set(self.user_name, self.password) self.client.connect(self.server, self.port, self.keepalive) except Exception as e : print_error_msg(str(e) + "mqtt_server: ip=%s, port = %d"%(self.server, self.port)) if self.client: # 如果self.client已经被初始化了，则断开连接 self.client.disconnect() self.client = None return None # 添加返回语句，告知调用者连接失败 return self.client def close(self): self.stop_request = True def publish(self, topic, message) : if self.is_connect : self.publish_list_lock.acquire() self.publish_list.append([topic, message]) self.publish_list_lock.release() self.condition.acquire() self.condition.notify() self.condition.release() return def publish_wait(self, topic, message, timeout) : if self.is_connect : self.publish_list_lock.acquire() self.publish_list.insert(0, [topic, message]) self.publish_list_lock.release() self.condition.acquire() self.condition.notify() self.condition.release() return def run(self): while self.stop_request == False : print_warning_msg("mqtt 线程启动, 服务器端口:server = %s:%d"%(self.server, self.port)) wait_open_timeout = 0 self.subscribe_list = [] #取消订阅列表, 连接后重新订阅 while self.client == None and self.stop_request == False: if wait_open_timeout == 0 : print_warning_msg("mqtt连接中, 服务器端口:server = %s:%d"%(self.server, self.port)) self.open() if self.client == None : time.sleep(0.1) wait_open_timeout += 0.1 if wait_open_timeout >= 5.0 : wait_open_timeout = 0 time.sleep(0.2) if self.client : self.client.loop_start() time.sleep(0.2) while self.stop_request == False and self.client.is_connected() : topic = None topic_message = None self.publish_list_lock.acquire() if len(self.publish_list) > 0 : topic_message = self.publish_list.pop(0) topic = topic_message[0] message = topic_message[1] self.publish_list_lock.release() if topic != None and message != None : self.client.publish(topic, message) self.condition.acquire() if len(self.publish_list) == 0: self.condition.wait(timeout = 0.1) self.condition.release() self.client.loop_stop() self.client.disconnect() self.client = None print_warning_msg("mqtt连接断开, 准备重启mqtt连接") print_error_msg("mqtt 线程停止") if __name__ == "__main__": object_mqtt1 = class_comm_mqtt_interface("object1 handler") object_mqtt2 = class_comm_mqtt_interface("object2 handler") global_mqtt_thread = class_comm_mqtt_thread() global_mqtt_thread.set_mqtt_server(server = "127.0.0.1", port = 1883, keep_alive = 60, user_name="admin", password="admin") global_mqtt_thread.add_unique_object("object1", object_mqtt1) global_mqtt_thread.add_unique_object("object2", object_mqtt2) global_mqtt_thread.start()