概述

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

信号

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

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

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

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

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

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_handler 和 notice_hanlder 对我喊的这个信号有意思，于是就触发了他们的逻辑，他们就会运行起来了，这就是所谓的信号。

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

login_signal.connect(log_handler)
login_signal.connect(notice_handler)

一样的效果。

订阅指定发布者

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

login_signal.connect(log_handler, sender=login)

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

弱连接

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

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/liuliqiang/blinker-examples

Reference

• Blinker Document