Python blinker 库教程
blinker 是一个用于实现信号机制的库，它提供了一种强大且灵活的方式来实现组件之间的松散耦合通信。信号机制允许一个对象（发送者）在某些事件发生时通知其他对象（接收者），而不需要彼此直接引用对方。
官方文档链接
安装 blinker 库
首先，确保你已经安装了 blinker 库。如果还没有安装，可以使用 pip 进行安装：
pip install blinker
使用 blinker 创建和发送信号
以下是如何使用 blinker 库创建和发送信号的基本示例。
1. 创建信号
使用 blinker.signal 创建一个信号：
from blinker import signal
# 创建一个信号
my_signal = signal('my-signal')
2. 连接接收者
定义一个接收者函数，并将其连接到信号。当信号被发送时，接收者函数将被调用：
def receiver(sender):
print(f"Received signal from {sender}")
# 连接接收者到信号
my_signal.connect(receiver)
3. 发送信号
使用 send 方法发送信号：
# 发送信号
my_signal.send('my-sender')
当你运行上述代码时，输出将会是：
Received signal from my-sender
信号发送和接收的更多用法
1. 发送带有数据的信号
你可以在发送信号时传递数据，这些数据会传递给所有接收者：
def receiver(sender, **kwargs):
print(f"Received signal from {sender} with data {kwargs}")
my_signal.connect(receiver)
# 发送信号并传递数据
my_signal.send('my-sender', foo='bar', baz=42)
输出：
Received signal from my-sender with data {'foo': 'bar', 'baz': 42}
2. 断开信号接收者
你可以使用 disconnect 方法断开信号接收者：
def receiver(sender):
print(f"Received signal from {sender}")
my_signal.connect(receiver)
# 发送信号
my_signal.send('my-sender')
# 断开接收者
my_signal.disconnect(receiver)
# 再次发送信号（接收者不会被调用）
my_signal.send('my-sender')
输出：
Received signal from my-sender
3. 使用信号装饰器
你可以使用装饰器将函数连接到信号：
@my_signal.connect
def receiver(sender):
print(f"Received signal from {sender}")
# 发送信号
my_signal.send('my-sender')
输出：
Received signal from my-sender
自定义信号命名空间
你可以使用 Namespace 类创建一个自定义的命名空间，以避免信号名称冲突：
from blinker import Namespace
# 创建一个命名空间
ns = Namespace()
# 在命名空间中创建信号
my_signal = ns.signal('my-signal')
def receiver(sender):
print(f"Received signal from {sender}")
my_signal.connect(receiver)
my_signal.send('my-sender')
输出：
Received signal from my-sender
信号中间件和过滤器
你可以在信号处理过程中添加中间件和过滤器，以实现更复杂的逻辑。
1. 信号中间件
信号中间件是在信号发送之前或之后执行的一段代码。你可以使用 connect_via 方法添加中间件。
def middleware(sender, **kwargs):
print(f"Middleware before signal from {sender}")
my_signal.connect(middleware)
def receiver(sender):
print(f"Received signal from {sender}")
my_signal.connect(receiver)
my_signal.send('my-sender')
输出：
Middleware before signal from my-sender
Received signal from my-sender
2. 信号过滤器
信号过滤器用于在信号处理过程中筛选或修改数据。你可以使用 filter 方法添加过滤器。
def filter_func(sender, **kwargs):
if kwargs.get('filter_out'):
return False
return True
my_signal.filter(filter_func)
def receiver(sender):
print(f"Received signal from {sender}")
my_signal.connect(receiver)
my_signal.send('my-sender', filter_out=True) # 不会被处理
my_signal.send('my-sender', filter_out=False) # 会被处理
输出：
Received signal from my-sender
使用示例
以下是一个综合示例，展示了如何使用 blinker 库创建和发送信号，连接接收者，并处理信号数据。
from blinker import signal
# 创建信号
user_logged_in = signal('user-logged-in')
# 定义接收者函数
def log_user_login(sender, **extra):
user = extra.get('user')
print(f"User {user} has logged in.")
# 连接接收者到信号
user_logged_in.connect(log_user_login)
# 发送信号并传递数据
user_logged_in.send('auth-system', user='Alice')
# 断开接收者
user_logged_in.disconnect(log_user_login)
# 再次发送信号（接收者不会被调用）
user_logged_in.send('auth-system', user='Bob')
输出：
User Alice has logged in.
总结
通过本教程，我们详细解析了 blinker 库的基本和高级功能，展示了如何使用该库创建和发送信号，连接和断开接收者，以及实现信号中间件和过滤器。信号机制是一种强大的设计模式，可以帮助开发者实现松散耦合的事件处理，提高代码的可维护性和扩展性。希望这篇教程对你有所帮助。更多详细信息和示例请参考 官方文档。
