Python模拟MQTT v3.1.1服务器

**Python 模拟 MQTT v3.1.1服务器**

MQTT (Message Queuing Telemetry Transport) 是一种轻量级的机器间通信协议，广泛应用于物联网、工业自动化等领域。下面我们将使用 Python 来模拟一个 MQTT v3.1.1服务器。

**依赖库**

本示例中，我们将使用以下库：

* `paho-mqtt`：用于 MQTT 协议的实现* `threading`：用于多线程处理首先，需要安装这些库：

bashpip install paho-mqtt threading

import threadingclass MQTTServer:
 def __init__(self):
 self.clients = {} # 存储客户端信息 self.topics = {} # 存储主题订阅信息 self.lock = threading.Lock() # 锁对象用于同步访问 def connect(self, client_id):
 """
 客户端连接服务器 :param client_id: 客户端ID :return: 是否成功连接 """
 with self.lock:
 if client_id in self.clients:
 return False # 如果客户端已经存在，则不允许重复连接 self.clients[client_id] = {} # 初始化客户端信息 return True def disconnect(self, client_id):
 """
 客户端断开连接 :param client_id: 客户端ID :return: 是否成功断开 """
 with self.lock:
 if client_id not in self.clients:
 return False # 如果客户端不存在，则不允许断开 del self.clients[client_id] # 删除客户端信息 return True def subscribe(self, client_id, topic):
 """
 客户端订阅主题 :param client_id: 客户端ID :param topic: 主题名称 :return: 是否成功订阅 """
 with self.lock:
 if client_id not in self.clients:
 return False # 如果客户端不存在，则不允许订阅 if topic in self.topics:
 self.topics[topic].append(client_id) # 添加主题订阅信息 return True self.topics[topic] = [client_id] # 初始化主题订阅信息 return True def unsubscribe(self, client_id, topic):
 """
 客户端取消订阅主题 :param client_id: 客户端ID :param topic: 主题名称 :return: 是否成功取消订阅 """
 with self.lock:
 if client_id not in self.clients or topic not in self.topics:
 return False # 如果客户端或主题不存在，则不允许取消订阅 self.topics[topic].remove(client_id) # 删除主题订阅信息 if len(self.topics[topic]) ==0: # 如果主题无任何订阅者，则删除主题 del self.topics[topic]
 return True def publish(self, topic, message):
 """
 发布消息到指定主题 :param topic: 主题名称 :param message: 消息内容 :return: 是否成功发布 """
 with self.lock:
 if topic not in self.topics or len(self.topics[topic]) ==0:
 return False # 如果主题不存在或无任何订阅者，则不允许发布 for client_id in self.topics[topic]:
 print(f"Publishing message to {client_id} on topic {topic}: {message}") # 模拟消息发送 return True

class MQTTClient:
 def __init__(self, client_id):
 self.client_id = client_id # 客户端ID def connect(self, server):
 """
 与服务器建立连接 :param server:服务器对象 :return: 是否成功连接 """
 return server.connect(self.client_id)

 def disconnect(self, server):
 """
 断开与服务器的连接 :param server:服务器对象 :return: 是否成功断开 """
 return server.disconnect(self.client_id)

 def subscribe(self, server, topic):
 """
 订阅主题 :param server:服务器对象 :param topic: 主题名称 :return: 是否成功订阅 """
 return server.subscribe(self.client_id, topic)

 def unsubscribe(self, server, topic):
 """
 取消订阅主题 :param server:服务器对象 :param topic: 主题名称 :return: 是否成功取消订阅 """
 return server.unsubscribe(self.client_id, topic)

 def publish(self, server, topic, message):
 """
 发布消息到指定主题 :param server:服务器对象 :param topic: 主题名称 :param message: 消息内容 :return: 是否成功发布 """
 return server.publish(topic, message)

if __name__ == "__main__":
 # 创建 MQTT服务器 server = MQTTServer()

 # 创建客户端 client1 = MQTTClient("client1")
 client2 = MQTTClient("client2")

 # 客户端连接服务器 print(client1.connect(server)) # True print(client2.connect(server)) # True # 客户端订阅主题 print(client1.subscribe(server, "topic1")) # True print(client2.subscribe(server, "topic1")) # True # 客户端发布消息 print(client1.publish(server, "topic1", "Hello, world!")) # True # 客户端取消订阅主题 print(client1.unsubscribe(server, "topic1")) # True print(client2.unsubscribe(server, "topic1")) # True # 客户端断开连接 print(client1.disconnect(server)) # True print(client2.disconnect(server)) # True