如何实现类Django Signals的观察者模式？代码问题求助

需求背景 #

需要让可观测对象在达到计算里程碑时发送中间值，参考Django Signals的观察者模式实现解耦，通过装饰器完成观察者与被观察者的绑定，最终实现信号子类调用所有装饰函数的功能（示例中Child同时作为信号类和sender简化场景）。

原代码的核心问题 #

1. 装饰器参数错误：原代码中mem_dec接收的是Child类而非实例，而信号的connect需要绑定到具体的信号实例上
2. 观察者存储错误：Signal.connect方法存储了函数名字符串，而非函数对象，导致send时无法直接调用函数
3. send方法逻辑倒置：原代码在send中调用mem_dec，完全搞反了注册与触发的逻辑，send的职责应该是触发已注册的观察者
4. 实例绑定混乱：观察者绑定到Child类而非实例，导致不同信号实例无法共享或独立管理观察者

修正后的完整实现 #

1. 信号模块（signals.py） #

class Signal:
    def __init__(self):
        # 存储观察者函数对象，而非名称
        self.observers = []

    def connect(self, receiver):
        # 避免重复注册同一观察者
        if receiver not in self.observers:
            self.observers.append(receiver)

    def send(self, sender, **named):
        # 触发所有已注册的观察者，返回调用结果
        results = []
        for receiver in self.observers:
            try:
                # 传递sender和关键字参数给观察者
                result = receiver(sender=sender, **named)
                results.append((receiver.__name__, result))
            except Exception as e:
                # 捕获异常，避免单个观察者报错影响全局
                results.append((receiver.__name__, str(e)))
        return results


class Child(Signal):
    # 维护一个全局实例，简化装饰器绑定（类似Django信号的单例用法）
    _instance = None

    def __init__(self):
        super().__init__()
        Child._instance = self

    @classmethod
    def get_instance(cls):
        if not cls._instance:
            cls()
        return cls._instance


def mem_dec(signal_instance=None, **kwargs):
    # 默认使用Child的全局实例，也支持传入自定义信号实例
    if signal_instance is None:
        signal_instance = Child.get_instance()

    def wrapper(func):
        # 将函数注册到信号实例
        signal_instance.connect(func)
        return func

    return wrapper


if __name__ == "__main__":
    # 先导入观察者，确保函数被注册
    from . import observers
    # 获取信号实例并发送信号
    send_signal = Child.get_instance()
    results = send_signal.send(sender=Child, a="henry")
    print("信号触发结果:", results)

2. 观察者模块（observers.py） #

from .signals import mem_dec

# 使用装饰器注册观察者函数，默认绑定Child的全局实例
@mem_dec()
def hello(sender, a, **kwargs):
    print(f'hey {a}, 发送者是 {sender.__name__}')
    return f"响应内容: {a}"

# 可以注册多个观察者
@mem_dec()
def log_info(sender, **kwargs):
    print(f'记录日志: sender={sender.__name__}, 参数={kwargs}')
    return "日志记录完成"

运行验证 #

执行signals.py的主函数，会输出：

hey henry, 发送者是 Child
记录日志: sender=Child, 参数={'a': 'henry'}
信号触发结果: [('hello', '响应内容: henry'), ('log_info', '日志记录完成')]

关键修正说明 #

• 观察者存储：改为存储函数对象，确保send时能直接调用
• 信号实例管理：通过全局实例简化装饰器调用，也支持自定义实例实现多信号隔离
• 触发逻辑：send方法直接遍历观察者列表并调用，传递正确的参数
• 异常处理：捕获观察者函数的异常，避免单个函数报错导致整个信号触发失败

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