Python 強(qiáng)大的信號庫 blinker 入門教程
1 信號
信號是一種通知或者說通信的方式，信號分為發(fā)送方和接收方。發(fā)送方發(fā)送一種信號，接收方收到信號的進(jìn)程會跳入信號處理函數(shù)，執(zhí)行完后再跳回原來的位置繼續(xù)執(zhí)行。
常見的 Linux 中的信號，通過鍵盤輸入 Ctrl+C，就是發(fā)送給系統(tǒng)一個(gè)信號，告訴系統(tǒng)退出當(dāng)前進(jìn)程。
信號的特點(diǎn)就是發(fā)送端通知訂閱者發(fā)生了什么。使用信號分為 3 步：定義信號，監(jiān)聽信號，發(fā)送信號。
Python 中提供了信號概念的通信模塊，就是blinker。
Blinker 是一個(gè)基于 Python 的強(qiáng)大的信號庫，它既支持簡單的點(diǎn)對點(diǎn)通信，也支持點(diǎn)對多點(diǎn)的組播。Flask 的信號機(jī)制就是基于它建立的。Blinker 的內(nèi)核雖然小巧，但是功能卻非常強(qiáng)大，它支持以下特性：
支持注冊全局命名信號 支持匿名信號 支持自定義命名信號 支持與接收者之間的持久連接與短暫連接 通過弱引用實(shí)現(xiàn)與接收者之間的自動斷開連接 支持發(fā)送任意大小的數(shù)據(jù) 支持收集信號接收者的返回值 線程安全
2 blinker 使用
安裝方法：
pip?install?blinker
2.1 命名信號
from?blinker?import?signal
#?定義一個(gè)信號
s?=?signal('king')
def?animal(args):
????print('我是小鉆風(fēng)，大王回來了，我要去巡山')
#?信號注冊一個(gè)接收者
s.connect(animal)
if?"__main__"?==?__name__:
????#?發(fā)送信號
????s.send()
2.2 匿名信號
blinker 也支持匿名信號，就是不需要指定一個(gè)具體的信號值。創(chuàng)建的每一個(gè)匿名信號都是互相獨(dú)立的。
from?blinker?import?Signal
s?=?Signal()
def?animal(sender):
????print('我是小鉆風(fēng)，大王回來了，我要去巡山')
s.connect(animal)
if?"__main__"?==?__name__:
????s.send()
2.3 組播信號
組播信號是比較能體現(xiàn)出信號優(yōu)點(diǎn)的特征。多個(gè)接收者注冊到信號上，發(fā)送者只需要發(fā)送一次就能傳遞信息到多個(gè)接收者。
from?blinker?import?signal
s?=?signal('king')
def?animal_one(args):
????print(f'我是小鉆風(fēng)，今天的口號是:?{args}')
def?animal_two(args):
????print(f'我是大鉆風(fēng)，今天的口號是:?{args}')
s.connect(animal_one)
s.connect(animal_two)
if?"__main__"?==?__name__:
????s.send('大王叫我來巡山，抓個(gè)和尚做晚餐！')
2.4 接收方訂閱主題
接受方支持訂閱指定的主題，只有當(dāng)指定的主題發(fā)送消息時(shí)才發(fā)送給接收方。這種方法很好的區(qū)分了不同的主題。
from?blinker?import?signal
s?=?signal('king')
def?animal(args):
????print(f'我是小鉆風(fēng)，{args}?是我大哥')
s.connect(animal,?sender='大象')
if?"__main__"?==?__name__:
????for?i?in?['獅子',?'大象',?'大鵬']:
????????s.send(i)
2.5 裝飾器用法
除了可以函數(shù)注冊之外還有更簡單的信號注冊方法，那就是裝飾器。
from?blinker?import?signal
s?=?signal('king')
@s.connect
def?animal_one(args):
????print(f'我是小鉆風(fēng)，今天的口號是:?{args}')
@s.connect
def?animal_two(args):
????print(f'我是大鉆風(fēng)，今天的口號是:?{args}')
if?"__main__"?==?__name__:
????s.send('大王叫我來巡山，抓個(gè)和尚做晚餐！')
2.6 可訂閱主題的裝飾器
connect的注冊方法用著裝飾器時(shí)有一個(gè)弊端就是不能夠訂閱主題，所以有更高級的connect_via方法支持訂閱主題。
from?blinker?import?signal
s?=?signal('king')
@s.connect_via('大象')
def?animal(args):
????print(f'我是小鉆風(fēng)，{args}?是我大哥')
if?"__main__"?==?__name__:
????for?i?in?['獅子',?'大象',?'大鵬']:
????????s.send(i)
2.7 檢查信號是否有接收者
如果對于一個(gè)發(fā)送者發(fā)送消息前要準(zhǔn)備的耗時(shí)很長，為了避免沒有接收者導(dǎo)致浪費(fèi)性能的情況，所以可以先檢查某一個(gè)信號是否有接收者，在確定有接收者的情況下才發(fā)送，做到精確。
from?blinker?import?signal
s?=?signal('king')
q?=?signal('queue')
def?animal(sender):
????print('我是小鉆風(fēng)，大王回來了，我要去巡山')
s.connect(animal)
if?"__main__"?==?__name__:
????
????res?=?s.receivers
????print(res)
????if?res:
????????s.send()
????
????res?=?q.receivers
????print(res)
????if?res:
????????q.send()
????else:
????????print("孩兒們都出去巡山了")
{4511880240:?0x10d02ae80;?to?'function'?at?0x10cedd430?(animal)>}
我是小鉆風(fēng)，大王回來了，我要去巡山
{}
孩兒們都出去巡山了
2.8 檢查訂閱者是否訂閱了某個(gè)信號
也可以檢查訂閱者是否由某一個(gè)信號
from?blinker?import?signal
s?=?signal('king')
q?=?signal('queue')
def?animal(sender):
????print('我是小鉆風(fēng)，大王回來了，我要去巡山')
s.connect(animal)
if?"__main__"?==?__name__:
????
????res?=?s.has_receivers_for(animal)
????print(res)
????res?=?q.has_receivers_for(animal)
????print(res)
True
False
3 基于 blinker 的 Flask 信號
Flask 集成 blinker 作為解耦應(yīng)用的解決方案。在 Flask 中，信號的使用場景如：請求到來之前，請求結(jié)束之后。同時(shí) Flask 也支持自定義信號。
3.1 簡單 Flask demo
from?flask?import?Flask
app?=?Flask(__name__)
@app.route('/',methods=['GET','POST'],endpoint='index')
def?index():
????return?'hello?blinker'
if?__name__?==?'__main__':
????app.run()
訪問127.0.0.1:5000時(shí)，返回給瀏覽器hello blinker。
3.2 自定義信號
因?yàn)?Flask 集成了信號，所以在 Flask 中使用信號時(shí)從 Flask 中引入。
from?flask?import?Flask
from?flask.signals?import?_signals
app?=?Flask(__name__)
s?=?_signals.singal('msg')
def?QQ(args):
????print('you?have?msg?from?QQ')
s.connect(QQ)
@app.route('/',methods=['GET','POST'],endpoint='index')
def?index():
????s.send()
????return?'hello?blinker'
if?__name__?==?'__main__':
????app.run()
3.3 Flask自帶信號
在 Flask 中除了可以自定義信號，還可以使用自帶信號。Flask 中自帶的信號有很多種，具體如下：
請求
request_started?=?_signals.signal('request-started')????????????????#?請求到來前執(zhí)行
request_finished?=?_signals.signal('request-finished')??????????????#?請求結(jié)束后執(zhí)行
?
模板渲染
before_render_template?=?_signals.signal('before-render-template')??#?模板渲染前執(zhí)行
template_rendered?=?_signals.signal('template-rendered')????????????#?模板渲染后執(zhí)行
?
請求執(zhí)行
got_request_exception?=?_signals.signal('got-request-exception')????#?請求執(zhí)行出現(xiàn)異常時(shí)執(zhí)行
request_tearing_down?=?_signals.signal('request-tearing-down')??????#?請求執(zhí)行完畢后自動執(zhí)行（無論成功與否）
appcontext_tearing_down?=?_signals.signal('appcontext-tearing-down')?#?請求上下文執(zhí)行完畢后自動執(zhí)行（無論成功與否）
?
請求上下文中
appcontext_pushed?=?_signals.signal('appcontext-pushed')????????????#?請求上下文push時(shí)執(zhí)行
appcontext_popped?=?_signals.signal('appcontext-popped')????????????#?請求上下文pop時(shí)執(zhí)行
message_flashed?=?_signals.signal('message-flashed')????????????????#?調(diào)用flask在其中添加數(shù)據(jù)時(shí)，自動觸發(fā)
下面以請求到來之前為例，看 Flask 中信號如何使用
from?flask?import?Flask
from?flask.signals?import?_signals,?request_started
import?time
app?=?Flask(__name__)
def?wechat(args):
????print('you?have?msg?from?wechat')
#?從flask中引入已經(jīng)定好的信號，注冊一個(gè)函數(shù)
request_started.connect(wechat)
@app.route('/',methods=['GET','POST'],endpoint='index')
def?index():
????return?'hello?blinker'
if?__name__?==?'__main__':
????app.run()
當(dāng)請求到來時(shí)，F(xiàn)lask 會經(jīng)過request_started 通知接受方，就是函數(shù)wechat，這時(shí)wechat函數(shù)先執(zhí)行，然后才返回結(jié)果給瀏覽器。
但這種使用方法并不是很地道，因?yàn)樾盘柌⒉恢С之惒椒椒ǎ酝ǔＴ谏a(chǎn)環(huán)境中信號的接收者都是配置異步執(zhí)行的框架，如 Python 中大名鼎鼎的異步框架 celery。
4 總結(jié)
信號的優(yōu)點(diǎn)：
解耦應(yīng)用：將串行運(yùn)行的耦合應(yīng)用分解為多級執(zhí)行 發(fā)布訂閱者：減少調(diào)用者的使用，一次調(diào)用通知多個(gè)訂閱者
信號的缺點(diǎn)：
不支持異步 支持訂閱主題的能力有限
還不過癮？試試它們