blinker 是 python 语言中一个处理信号的库,对比 python 语言中自带的信号库,blinker 提供了一些非常有用的特性,例如命名空间、线程安全以及指定发送者等,都是默认信号库所不具备的,本文将对这些特性进行分解,解释一下一些特性的意思,然后再给出一些例子进行尝试。

信号

如果一来就谈各种信号好像不太合适,因为可能有人不太清楚信号是啥子,信号有什么用,所以,开头是以一个简单的讲解信号的例子开始,说明一下什么是信息,这里先假设一个场景,就举登陆的例子吧,假如我们对账号进行审计,也就是说每次有人登陆,我们都需要进行记录,那么可能最冲动的做法就是这样来:

1
2
3
def login(username, password):
    ... ... # 各种和登陆有关的日志
    log_login(username)

也就是直接在 login 的地方加上记录的函数,这样呢是可以达到我们的目的,那么如果以后又多了一个需求,不仅仅要记录登陆的情况,对于异地IP登陆的账号我们可能还需要进行提醒用户,让用户知道自己的账号是不是被盗用了,那么这个时候我们好像得这么改:

1
2
3
4
5
def login(username, password):
    ... ... # 各种和登陆有关的日志
    log_login(username)
    if need_notice(username):
        notice_user(username)

随着需求的不断增加,我们的登陆函数越来越不纯净了,会塞上一大堆和登陆无关的代码,这个时候是时候来一波重构了。重构的办法有很多,那么我就以信号的方式来做一个实例,顺便阐述一下什么是信号:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def login(username, password):
    ... ... # 各种和登陆有关的日志
    login_signal.send(username)

@login_signal.connect()
def log_handler(username):
    ....

@login_signal.connect()
def notice_handler(username):
    if need_notice(username):
        notice_user(username)

这里在登陆的代码中就只加了一行,就是login_signal.send(username),其实这就是所谓的信号了,我大声对外喊: username登录了。然后,有两个人 log_handlernotice_hanlder 对我喊的这个信号有意思,于是就触发了他们的逻辑,他们就会运行起来了,这就是所谓的信号。

发送信号只需要简单调用 send 函数就可以了,同时,调用的时候可以任意传递参数,这些参数都会被感兴趣的人接收到并且利用起来。而怎么表示我对这个信号感兴趣呢?也很简单,这个例子中用了一个装饰器connect,除了 connect 还有 connect_via,不用装饰器也行,那就直接调用函数:

1
2
login_signal.connect(log_handler)
login_signal.connect(notice_handler)

一样的效果。

订阅指定发布者

刚才说了,我大声对外喊了一声 xxx登陆了,只为了让感兴趣的人知道,那么对于不感兴趣的人来说是不是有点吵到人家了,如果有一天我喊:我被人打了,那小明的妈妈听到了以为小明被打了岂不是很乌龙?所以这里引入了订阅指定发布者的功能,也就是说,我只听 xxx 喊出的话,其他人说的我都忽略,那么这在 blinker 中也是可以做到的,那就是通过指定 sender 参数:

1
login_signal.connect(log_handler, sender=login)

这样的话,log_handler 就只对 login 发出的 login_signal 感兴趣呢,对于其他的发出了,其实你喊破喉咙它也不会理。

弱连接

这里重新假设一种情况,那就是我约你去玩,前提是你已经在家外面的时候,我叫你去玩你才会陪我去玩,要是你在家里学习的话,我叫你去玩你也不会理我,那么 blinker 对于这种情况就引入了弱连接的概念,代码连接

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
def test_weak():
    @weak_signal.connect_via(Signal.ANY, weak=True)
    def receive_data(sender, **kw):
        print ("Caught weak signal from : {}, data: {}".format(sender, kw))
        return "test"

@weak_signal.connect_via(Signal.ANY, weak=True)
def receive_data2(sender, **kw):
    print ("Caught signal2 from : {}, data: {}".format(sender, kw))
    return "received2"

当你在外面调用 weak_signal 的时候,test_weak 内部的 receive_data 是不会理你的,因为它在家里学习,那么怎么在会理你呢,那只能 receive_data2 会理你,因为他在家外面玩耍,可以陪你玩。那怎么才能让 test_weak 里面的 receive_data 也陪你玩呢,那我觉得只能你去他家里陪ta 读书了,也就是在 test_weak 里面 send 一个 weak_signal,这时他就会响应了。

线程安全

对于 python 原生的 signal,它其实是进程内共享的,也就是对于多线程程序,如果一个线程发送了 KILL 命令,其他线程都会同时死掉,这样就很不好了。所以 blinker 避免了这个问题。

例子

github 地址:https://github.com/luke0922/blinker-examples

Reference