如何在Python中动态创建单例类以实现消息主题功能

核心实现思路 #

用类级别的缓存字典存储已创建的Topic实例，确保每个主题名称对应唯一单例；Topic类封装观察者的注册、移除和消息通知逻辑，GQueue.get_topic()方法负责实例的创建与缓存复用。

代码实现（Python） #

class Topic:
    def __init__(self, name):
        self.name = name
        self._observers = []

    def subscribe(self, observer):
        """注册观察者"""
        if observer not in self._observers:
            self._observers.append(observer)

    def unsubscribe(self, observer):
        """移除观察者"""
        if observer in self._observers:
            self._observers.remove(observer)

    def notify(self, message):
        """向所有观察者推送消息"""
        for observer in self._observers:
            observer.update(self.name, message)


class GQueue:
    # 类级别的缓存，存储已创建的Topic单例
    _topics = {}

    @classmethod
    def get_topic(cls, topic_name):
        """获取指定名称的Topic单例，不存在则创建"""
        if topic_name not in cls._topics:
            cls._topics[topic_name] = Topic(topic_name)
        return cls._topics[topic_name]


# 示例观察者类
class Observer:
    def __init__(self, name):
        self.name = name

    def update(self, topic_name, message):
        print(f"观察者{self.name}收到[{topic_name}]的消息: {message}")


# 用法示例
if __name__ == "__main__":
    # 获取动态生成的Topic单例
    topic_x = GQueue.get_topic("X")
    topic_y = GQueue.get_topic("Y")
    topic_x_again = GQueue.get_topic("X")

    # 验证单例：两次获取的X是同一个实例
    print(topic_x is topic_x_again)  # 输出: True

    # 创建观察者并订阅
    observer1 = Observer("A")
    observer2 = Observer("B")
    topic_x.subscribe(observer1)
    topic_x.subscribe(observer2)
    topic_y.subscribe(observer1)

    # 发送消息
    topic_x.notify("Hello Topic X")
    topic_y.notify("Hello Topic Y")

关键细节说明 #

• 单例保证：GQueue._topics作为全局缓存，每次调用get_topic()先检查缓存，只有当主题不存在时才创建新实例，确保同一名称的Topic仅初始化一次。
• 观察者管理：Topic类通过_observers列表维护订阅者，subscribe/unsubscribe方法处理注册和移除，notify方法遍历所有观察者触发消息回调。
• 动态扩展性：无需提前定义固定名称的Topic类，传入任意字符串作为主题名称，即可生成对应的单例实例，完全满足topic_x = GQueue.get_topic("X")的调用需求。

内容的提问来源于stack exchange，提问作者Prateek Trivedi