Python项目如何构建跨模块事件发布与订阅机制【教程】

推荐初学者用标准库weakref自定义事件总线，生产项目用blinker库；核心是解耦模块依赖，需防范循环发布、异常中断、生命周期错配等陷阱。

Python项目中实现跨模块事件发布与订阅，核心是解耦模块间直接调用，让一个模块“发消息”，其他模块“听消息”而不互相依赖。关键不在于造轮子，而在于选对轻量、可控、易调试的方案。

用标准库 weakref + 自定义事件总线（推荐初学者）

不依赖第三方，逻辑透明，适合中小项目或想理解底层机制的场景。本质是维护一个事件名到回调函数列表的映射，用 weakref 避免内存泄漏（防止订阅者被意外强引用导致无法回收）。

• 定义一个全局或单例的 EventBus 类，内部用 defaultdict(list) 存储事件名与弱引用回调
• subscribe(event_name, callback)：用 weakref.WeakMethod(callback) 或 weakref.ref(callback) 注册
• publish(event_name, *args, **kwargs)：遍历对应回调列表，调用前检查引用是否有效（cb_ref() is not None）
• 模块A只需 import 这个 EventBus 并 publish；模块B import 后 subscribe，彼此无导入依赖

用 blinker 库（推荐生产项目）

轻量、成熟、支持信号分组、临时订阅、自动清理，比手写更稳。安装：pip install blinker。

• 定义信号：user_registered = signal('user-registered')
• 模块B订阅：user_registered.connect(handle_user_registration, sender=ANY)
• 模块A触发：user_registered.send(app, user_id=123, email="a@b.com")
• 支持 sender 过滤（如只响应某类对象发出的信号）、临时连接（connect_via）、异步兼容（配合 asyncio.run_in_executor 等）

避免常见陷阱

事件机制看似简单，实际落地容易踩坑：

标签： python app 回调函数 ai ios 标准库 red