Blinker是Python生态里一个轻量级的信号库，整个库只有几百行代码，却是Flask、Django之外的很多Web框架实现事件机制的基础。它的核心思想很简单：发送方在某个时机发出一个信号，不关心谁在监听；接收方订阅自己感兴趣的信号，不关心是谁发出来的。两边通过信号名字关联，完全解耦。这篇文章就来详细讲讲Blinker的安装、核心API和实际项目中的用法。
一、Blinker的核心概念与基本用法
先安装库，直接用pip即可：
pip install blinker
Blinker的使用围绕三个动作展开：创建信号、订阅信号、发送信号。创建信号需要一个全局的命名空间对象，通常写成下面这样：
from blinker import signal
# 创建一个名为'order-completed'的信号
order_completed = signal('order-completed')
# 定义接收函数，第一个参数固定是发送者，后续是关键字参数
def notify_user(sender, **kwargs):
print(f'通知用户：订单 {kwargs.get("order_id")} 已完成')
# 订阅信号
order_completed.connect(notify_user)
# 发送信号，sender是必传参数
order_completed.send('checkout-module', order_id=10086)执行后控制台会打印出通知信息。这里有几个细节值得注意：send方法的第一个参数是sender，表示信号的发送者，这是Blinker强制要求的，通常传模块名、类名或者专门的哨兵对象。订阅函数的第一个参数也会接收到这个sender，其余数据全部通过关键字参数传递。
除了用connect方法订阅，Blinker还提供了更Pythonic的装饰器写法：
from blinker import signal
order_completed = signal('order-completed')
@order_completed.connect
def notify_user(sender, **kwargs):
print(f'装饰器方式订阅：{kwargs}')
order_completed.send('checkout', order_id=1)两种方式本质相同，装饰器写法省去了手动调用connect的步骤，代码更紧凑。另外还有一个connect_via装饰器，可以同时指定只监听某个特定sender发出的信号，后面会详细讲。
二、用信号机制解耦业务：一个订单处理的实例
信号机制最大的价值在于解耦。假设一个电商项目里，订单完成后需要做三件事：给用户发短信、给运营发统计报表、清理购物车缓存。如果都写在一个函数里，主业务逻辑会越来越臃肿，改动任何一件事都要动主流程。用Blinker改造后，主流程只负责发信号：
# services/order.py
from blinker import signal
order_completed = signal('order-completed')
def finish_order(order):
# 核心业务：更新订单状态
order.status = 'completed'
order.save()
# 只发信号，不关心后续处理
order_completed.send('order-service', order=order)
# services/notify.py
from services.order import order_completed
@order_completed.connect
def send_sms(sender, order, **kwargs):
print(f'给 {order.user.phone} 发送短信')
# services/report.py
from services.order import order_completed
@order_completed.connect
def generate_report(sender, order, **kwargs):
print(f'记录订单 {order.id} 到报表')这样拆分之后，主业务模块完全不知道短信和报表的存在。以后要新增一个积分奖励逻辑，只需要在新模块里写一个订阅函数，主流程一行代码都不用改，符合开闭原则。
需要注意接收函数的执行顺序。Blinker内部用一个集合存储订阅者，不保证回调的执行顺序，所以不要在多个订阅函数之间设计依赖关系。如果某个订阅函数抛出异常，会中断后续订阅函数的执行并向上传播，所以每个订阅函数内部最好做好异常处理，避免一个模块的问题影响其他模块。
三、sender过滤与弱引用：两个容易踩的坑
第一个要讲的是sender过滤。实际项目中，同一个信号可能由不同模块发出，而接收方往往只关心其中某个来源。这时可以用connect_via指定sender：
from blinker import signal
user_login = signal('user-login')
@user_login.connect_via('web')
def only_web_login(sender, **kwargs):
print(f'Web端登录：{kwargs}')
user_login.send('web', user_id=1) # 会触发回调
user_login.send('app', user_id=1) # 不会触发回调sender也可以传任意对象，比如类本身。Flask就是用app实例作为sender，让订阅方区分是哪个应用实例发出的信号。另外Blinker内置了一个ANY哨兵，connect时如果不指定sender，默认监听所有sender，等价于connect_via(signal.ANY)。
第二个坑是弱引用。Blinker默认对订阅函数持有弱引用，这意味着如果函数对象没有被其他地方引用，被垃圾回收后订阅会自动失效。最典型的翻车场景是把lambda或者临时函数直接传给connect：
from blinker import signal
sig = signal('demo')
sig.connect(lambda sender, **kw: print('执行了')) # lambda没有引用，随时可能被回收
sig.send('main') # 可能不会打印任何东西
# 正确做法：weak=False，或者给函数保存一个引用
def on_demo(sender, **kw):
print('执行了')
sig.connect(on_demo, weak=False)
sig.send('main') # 稳定触发一般建议模块级的具名函数不需要额外处理，因为模块会持有引用；只有在connect匿名函数、嵌套函数或者动态创建的回调时，才需要显式传weak=False，否则信号发出去没人响应，排查起来非常费劲。
四、临时订阅与信号的高级用法
Blinker还提供了一些进阶能力。比如临时订阅某段逻辑，用connected_to上下文管理器，退出作用域后自动取消订阅，适合在测试或事务场景中使用：
from blinker import signal
sig = signal('temp-demo')
def handler(sender, **kw):
print('临时处理器被触发')
with sig.connected_to(handler):
sig.send('main') # 处理器生效
sig.send('main') # 处理器已自动断开还可以给信号设置优先级。Blinker本身没有直接的优先级参数，但可以通过给接收函数设置priority属性并配合signal.receiver_connected来定制，更简单的做法是保持订阅逻辑无顺序依赖，从设计上规避问题。
最后一点，send方法返回一个列表，包含所有被触发的订阅函数及其返回值，可以用来确认信号是否真的有人处理：
results = sig.send('main')
for receiver, ret in results:
print(f'{receiver.__name__} 返回了 {ret}')这在调试信号订阅是否生效时特别有用。总体来说，Blinker适合的场景是主流程与后续扩展动作解耦，比如日志记录、缓存清理、消息推送等横切逻辑。如果你的项目里出现了大量回调注册代码，或者Flask扩展需要提供钩子能力，Blinker都是值得优先考虑的方案。它的源码很短，读懂它对理解Python的弱引用和描述符机制也很有帮助。