Python的信号库Blinker有何用法?

Blinker 是一个轻量级 Python 信号库,用于实现对象间解耦通信。1. 使用 pip install blinker 安装;2. 通过 Signal() 创建信号,connect() 绑定接收函数,send() 发送信号并触发回调;3. 可使用 signal('name') 创建命名信号,便于管理;4. 调用 disconnect() 可断开接收器;5. has_receivers_for() 检查是否有监听者;6. 支持 sender 参数过滤发送者,但需在回调中手动判断。适用于插件系统、事件通知等场景。

Blinker 是 Python 中一个快速、简单且轻量级的信号(事件)库,用于实现对象之间的解耦通信。它允许你定义“信号”,在程序运行过程中发送这些信号,并让预先注册的函数(接收器)对信号做出响应。这种机制常用于插件系统、事件监听、日志记录、状态变更通知等场景。

1. 安装 Blinker

使用 pip 安装 Blinker:

pip install blinker

2. 创建和使用自定义信号

Blinker 的核心是 Signal 类。你可以创建一个信号实例,然后连接回调函数来监听它。

示例:

from blinker import Signal

# 创建一个信号
user_logged_in = Signal()

# 定义接收函数
def notify_user(sender, **kwargs):
    print(f"用户 {sender} 已登录,附加信息: {kwargs}")

# 将函数连接到信号
user_logged_in.connect(notify_user)

# 发送信号
user_logged_in.send("alice", info="从Web端登录")

输出:

用户 alice 已登录,附加信息: {'info': '从Web端登录'}

3. 使用命名信号(blinker.namespace)

当信号较多时,可以用 Namespace 来组织管理。

示例:

from blinker import signal

# 创建命名信号空间
my_signals = signal('user_login')

# 第一次调用 signal() 创建信号,之后调用返回同一个实例
login_signal = signal('user_login')

def on_login(sender):
    print(f"{sender} 登录了")

login_signal.connect(on_login)
login_signal.send("bob")

4. 断开信号连接

可以随时断开某个接收器,防止其再次被调用。

示例:

def temp_handler(sender):
    print("临时处理")

login_signal.connect(temp_handler)
login_signal.send("test") # 触发
login_signal.disconnect(temp_handler) # 断开
login_signal.send("test2") # 不再触发

5. 检查信号是否被连接

可通过 has_receivers_for 判断是否有监听者。

if login_signal.has_receivers_for("any_sender"):
    print("有接收者")

6. 使用匿名信号与 sender 过滤

接收函数可以通过 sender 参数指定只接收特定发送者的信号。

示例:

admin_signal = Signal()

def only_for_admin(sender):
    if sender == "admin":
        print("管理员操作已通知")

admin_signal.connect(only_for_admin)
admin_signal.send("admin") # 触发
admin_signal.send("user") # 仍会调用,需在函数内判断

注意:Blinker 不自动过滤 sender,需在回调中手动判断。

基本上就这些。Blinker 虽小,但在需要事件通知又不想引入复杂框架时非常实用。