一、 概述
为了模块间解耦，消息总线是常用的方式。
在其它文章中分别提到了lua和C++语言的消息总线的实现
lua语言的消息总线的实现：lua简单消息总线实现，类似于C++ Qt的信号槽
cpp语言的消息总线的实现：C++的阻塞式消息总线message_bus实现
消息总线实现的基本原理如下：被通信对象向消息总线发布一个主题，这个主题包含消息主题和消息处理函数，消息主题标示某个特定的主题，消息处理函数用来响应该主题的某种消息类型。通信对象向消息总线发送某个特定主题和消息参数，总线就会根据消息主题和消息参数找到对应的消息处理函数处理该请求。
二、Python消息总线的实现
class PyBus (object):def __init__(self,):self.clear()def clear(self):self.subscriptions = {}def subscribe(self, subject, owner, func):if owner not in self.subscriptions:self.subscriptions[owner] = {}self.subscriptions[owner][subject] = funcdef has_subscription(self, owner, subject):return owner in self.subscriptions and subject in self.subscriptions[owner]def publish(self, subject, *args, **kwargs):for owner in self.subscriptions.keys():if self.has_subscription(owner, subject):self.subscriptions[owner][subject](*args, **kwargs)class BusSingleton(PyBus):def foo(self):passbus_singleton = PyBus()
核心是内部维护的subscriptions字典
三、测试代码
if __name__ == "__main__":START = 1111class Test1(object):def start(self):print("start1")class Test2(object):def start(self):print("start2")test1 = Test1()test2 = Test2()bus_singleton.subscribe(START, test1, test1.start)bus_singleton.subscribe(START, test2, test2.start)bus_singleton.publish(START)
