目录

一、全局管理者

二、broker内部广播

2.1 监听主题消息线程

2.2 客户端发送消息的实例方法

2.3 保存客户端实例

三、leader客户端的作用

3.1 leader指定方式

四、消息内容定义

五、客户端连接处理

2.1 连接处理

2.2 订阅消息处理

2.3 接收新的订阅消息

2.4 发布消息处理

2.5 接收新的发布消息

六、客户端断开处理

本篇主要阐述Broker的实现内容。

Broker作为一个管理订阅和发布的管理系统，它将记录订阅者，当有发布者发布相应主题的内容时，将该主题内容分发给订阅者。

因此在Broker中需要一个全局保存连接上来的客户端，当客户端为订阅者，就将订阅者订阅的主题保存下来。

一、全局管理者

class GlobalHandle:
    def __init__(self):
        self.client_dic = {} # 客户端保存对象
        self.send_msg_queue = queue.Queue() # 给订阅者们发数据通道

        self.lock = threading.Lock() # 通常使用的lock

        self.leader_lock = threading.Lock() # leader使用的lock
        self.leader_client = None # leader客户端

        super().__init__()


GLOBAL_HANDLE = GlobalHandle() # 将管理者作为单例使用

全局管理者主要拥有以上变量，它作为一个单例去使用：

• client_dic：将保存所有连接broker的客户端；
• send_msg_queue：当订阅者订阅相关主题，只要有一个发布者发布主题消息，该通道将插入一条主题消息用来广播给所有订阅者；
• leader_client：当broker被关闭时，可通过leader客户端去重新启动broker系统，该变量作为leader的标识符。

上面除了client_dic比较好理解，其他变量将会进一步解释。

二、broker内部广播

当有一个主题发布上来，broker需要在系统内部将这条消息广播给所有的订阅者：

我这边使用send_msg_queue加一个线程来实现数据的广播：

1. 发布者发布主题消息，放入send_msg_queue管道；
2. 线程拉取send_msg_queue管道数据；
3. 获取所有客户端中订阅该主题的客户端；
4. 调用客户端的发送处理方法，将消息广播到客户端中。

获取所有客户端的方法很简单，client_dic中就已经保存了，而每个客户端的发送处理方法则需要进入到每个客户端的连接线程中去获取，因此我需要通过连接server时，将客户端的发送方法获取，所以我自己构建了一个socketserver去实现：

class MQServer(TCPServer):
    allow_reuse_address = True # 在linux系统中，该server可以立即重启

    def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True, radio_func_dic={}):
        self.radio_func_dic = radio_func_dic # 存储客户端的发送方法
        TCPServer.__init__(self, server_address, RequestHandlerClass, bind_and_activate=bind_and_activate)


class MQThreadingTCPServer(ThreadingMixIn, MQServer):
    pass

该server中有一个radio_func_dic，当客户端连接成功时，会将自己的实例保存到对象中，然后我们通个对象就可以调取该实例的所有方法。

2.1 监听主题消息线程

def global_radio_handle(radio_func_dic):
    """
    监听主题消息线程
    :param radio_func_dic: 保存客户端实例的对象
    :return: 
    """
    global GLOBAL_HANDLE # 获取全局管理者

    while True:
        if not GLOBAL_HANDLE.send_msg_queue.empty(): # 当管道不为空
            res = GLOBAL_HANDLE.send_msg_queue.get() # 获取管道消息
            key = res.get('key') # 消息主题
            mq_type = res.get("type") # 消息类型

            with GLOBAL_HANDLE.lock:
                for ip_address in radio_func_dic: # 如果客户端实例存在
                    if ip_address in GLOBAL_HANDLE.client_dic: # 如果客户端存在
                        if mq_type == "leader": # 消息主题为leader
                            if radio_func_dic[GLOBAL_HANDLE.leader_client]:
                                radio_func_dic[GLOBAL_HANDLE.leader_client].send_msg_func(res) # 发送leader消息
                        elif key in GLOBAL_HANDLE.client_dic[ip_address]: # 客户端如果订阅了主题
                            if radio_func_dic[ip_address]:
                                radio_func_dic[ip_address].send_msg_func(res) # 给客户端发送主题消息

2.2 客户端发送消息的实例方法

    def send_msg_func(self, res):
        """
        发送消息给客户端
        :param res: 消息对象
        :return: 
        """
        res = json.dumps(res)
        res = res.encode()
        try:
            self.request.send(res)
        except:
            pass

2.3 保存客户端实例

# 当客户端连接成功，保存客户端实例
with self.global_handle.lock:
    self.server.radio_func_dic[self.ip_address] = self

现在，如果有发布者发布主题消息时，我们broker就可以将该主题消息发送给相应的订阅者。

三、leader客户端的作用

为了达到broker被关闭时，我们的订阅者或发布者去启动我们的broker时，只有一个客户端去启动，我们需要指定一个leader，如果不这么做，我们的客户端一旦检测到broker未运行，将大批量去同时启动broker，这样不光会消耗不必要的性能也会造成一定的麻烦，例如：在windows下可能会启动多个野broker程序。

3.1 leader指定方式

1. 客户端连接到broker，记录客户端；
2. 判断当前leader是否存在，如果不存在，指定当前客户端为leader，将leader指定消息发送给客户端；
3. 当客户端断开连接，查看当前客户端是否为leader，如果为leader，将leader字段置空，并从当前活跃的客户端中指定第一个客户端为leader，将leader指定消息发送给客户端。

当leader为空，第一台客户端连接broker时指定leader：

with self.global_handle.leader_lock:
    if not self.global_handle.leader_client: # leader不存在
        data = {"type": "leader"}
        self.global_handle.leader_client = self.ip_address # 指定leader
        self.send_msg_func(data) # 将指定的消息发送给客户端

当客户端结束时，重新指定leader：

with self.global_handle.leader_lock:
    if self.global_handle.leader_client == self.ip_address: # 当前断开的客户端为leader
        self.global_handle.leader_client = None # 重置leader

    for key in self.server.radio_func_dic: # 获取当前活跃客户端
        self.global_handle.leader_client = key # 指定第一个客户端
        data = {"type": "leader"}
        self.global_handle.send_msg_queue.put(data) # 去让指定客户端发送leader消息
        break

客户端收到{"type": "leader"}的leader指定消息时，会将客户端内的is_leader变量设置为True，这样当broker被关闭时，将仅有一台客户端去重启broker。

四、消息内容定义

订阅消息：{"type": "sub", "key": "主题", "data": "xxx"}

发布消息：{"type": "pub", "key": "主题", "data": "xxx"}

leader消息：{"type": "leader"}

五、客户端连接处理

2.1 连接处理

    def handle(self):
        res = None
        try:
            res = self.request.recv(2048)
        except:
            pass

        if res:
            if res == b"PyMQ test": # 用来测试broker运行
                self.request.send(b"Yes")
            else:
                try:
                    res = json.loads(res.decode())
                except:
                    return

                mq_type = res.get("type")
                if mq_type:
                    if self.ip_address not in self.global_handle.client_dic: # 将客户端加入到client_dic中
                        with self.global_handle.lock:
                            self.global_handle.client_dic[self.ip_address] = {}

                    with self.global_handle.leader_lock:
                        if not self.global_handle.leader_client: # 如果leader不存在
                            data = {"type": "leader"}
                            self.global_handle.leader_client = self.ip_address # 指定leader
                            self.send_msg_func(data) # 发送leader信息

                    if mq_type == "sub":
                        self.__sub_handle(res) # 进行订阅方法

                        self.receive_sub_thread = threading.Thread(target=self.__receive_sub_handle, args=())
                        self.receive_sub_thread.start()
                    elif mq_type == "pub":
                        self.__pub_handle(res) # 进行发布方法

                        self.receive_pub_thread = threading.Thread(target=self.__receive_pub_handle, args=())
                        self.receive_pub_thread.start()

                    with self.global_handle.lock:
                        self.server.radio_func_dic[self.ip_address] = self # 保存客户端实例

                    while True: # 保活
                        if self.receive_sub_thread and not self.receive_sub_thread.is_alive():
                            break
                        if self.receive_pub_thread and not self.receive_pub_thread.is_alive():
                            break

                        time.sleep(10)

                    self.event.set() # 将所有线程退出

2.2 订阅消息处理

    def __sub_handle(self, res):
        key = res.get("key") # 获取主题

        if key:
            if key not in self.global_handle.client_dic[self.ip_address]: # 如果主题未存在于客户端的订阅中，则将该主题赋予该客户端
                with self.global_handle.lock:
                    self.global_handle.client_dic[self.ip_address][key] = 1 # 赋予客户端主题

2.3 接收新的订阅消息

    def __receive_sub_handle(self):
        while not self.event.is_set():
            try:
                res = self.request.recv(2048)
            except:
                break

            try:
                res = json.loads(res.decode())
                mq_type = res.get("type")
                if mq_type:
                    if mq_type == "sub":
                        self.__sub_handle(res) # 将新的订阅进行存储
            except:
                pass

2.4 发布消息处理

    def __pub_handle(self, res):
        key = res.get("key")

        if key:
            if "value" in res:
                self.global_handle.send_msg_queue.put(res) # 将主题消息广播出去

2.5 接收新的发布消息

    def __receive_pub_handle(self):
        while not self.event.is_set():
            try:
                res = self.request.recv(2048)
            except:
                break

            try:
                res = json.loads(res.decode())
                mq_type = res.get("type")
                if mq_type:
                    if mq_type == "pub":
                        self.__pub_handle(res) # 将发布消息进行处理
            except:
                pass

六、客户端断开处理

    def finish(self):
        try:
            with self.global_handle.lock:
                self.server.radio_func_dic[self.ip_address] = None # 清除当前客户端实例
                del self.server.radio_func_dic[self.ip_address]
        except:
            pass

        try:
            with self.global_handle.lock:
                del self.global_handle.client_dic[self.ip_address] # 清除当前客户端
        except:
            pass

        with self.global_handle.leader_lock:
            if self.global_handle.leader_client == self.ip_address:
                self.global_handle.leader_client = None

                for key in self.server.radio_func_dic:
                    self.global_handle.leader_client = key
                    data = {"type": "leader"}
                    self.global_handle.send_msg_queue.put(data)
                    break

未完待续。