Python 強(qiáng)大的信號(hào)庫(kù) blinker 入

安裝方法：

pip install blinker2.1 命名信號(hào)

from blinker import signal# 定義一個(gè)信號(hào)s = signal('king')def animal(args): print('我是小鉆風(fēng)，大王回來(lái)了，我要去巡山')# 信號(hào)注冊(cè)一個(gè)接收者s.connect(animal)if "__main__" == __name__: # 發(fā)送信號(hào) s.send()2.2 匿名信號(hào)

blinker 也支持匿名信號(hào)，就是不需要指定一個(gè)具體得信號(hào)值。創(chuàng)建得每一個(gè)匿名信號(hào)都是互相獨(dú)立得。

from blinker import Signals = Signal()def animal(sender): print('我是小鉆風(fēng)，大王回來(lái)了，我要去巡山')s.connect(animal)if "__main__" == __name__: s.send()2.3 組播信號(hào)

組播信號(hào)是比較能體現(xiàn)出信號(hào)優(yōu)點(diǎn)得特征。多個(gè)接收者注冊(cè)到信號(hào)上，發(fā)送者只需要發(fā)送一次就能傳遞信息到多個(gè)接收者。

from blinker import signals = signal('king')def animal_one(args): print(f'我是小鉆風(fēng)，今天得口號(hào)是: {args}')def animal_two(args): print(f'我是大鉆風(fēng)，今天得口號(hào)是: {args}')s.connect(animal_one)s.connect(animal_two)if "__main__" == __name__: s.send('大王叫我來(lái)巡山，抓個(gè)和尚做晚餐！')2.4 接收方訂閱主題

接受方支持訂閱指定得主題，只有當(dāng)指定得主題發(fā)送消息時(shí)才發(fā)送給接收方。這種方法很好得區(qū)分了不同得主題。

from blinker import signals = 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ù)注冊(cè)之外還有更簡(jiǎn)單得信號(hào)注冊(cè)方法，那就是裝飾器。

from blinker import signals = signal('king')等s.connectdef animal_one(args): print(f'我是小鉆風(fēng)，今天得口號(hào)是: {args}')等s.connectdef animal_two(args): print(f'我是大鉆風(fēng)，今天得口號(hào)是: {args}')if "__main__" == __name__: s.send('大王叫我來(lái)巡山，抓個(gè)和尚做晚餐！')2.6 可訂閱主題得裝飾器

connect得注冊(cè)方法用著裝飾器時(shí)有一個(gè)弊端就是不能夠訂閱主題，所以有更高級(jí)得connect_via方法支持訂閱主題。

from blinker import signals = signal('king')等s.connect_via('大象')def animal(args): print(f'我是小鉆風(fēng)，{args} 是我大哥')if "__main__" == __name__: for i in ['獅子', '大象', '大鵬']: s.send(i)2.7 檢查信號(hào)是否有接收者

如果對(duì)于一個(gè)發(fā)送者發(fā)送消息前要準(zhǔn)備得耗時(shí)很長(zhǎng)，為了避免沒(méi)有接收者導(dǎo)致浪費(fèi)性能得情況，所以可以先檢查某一個(gè)信號(hào)是否有接收者，在確定有接收者得情況下才發(fā)送，做到精確。

from blinker import signals = signal('king')q = signal('queue')def animal(sender): print('我是小鉆風(fēng)，大王回來(lái)了，我要去巡山')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: <weakref at 0x10d02ae80; to 'function' at 0x10cedd430 (animal)>}我是小鉆風(fēng)，大王回來(lái)了，我要去巡山{}孩兒們都出去巡山了2.8 檢查訂閱者是否訂閱了某個(gè)信號(hào)

也可以檢查訂閱者是否由某一個(gè)信號(hào)

from blinker import signals = signal('king')q = signal('queue')def animal(sender): print('我是小鉆風(fēng)，大王回來(lái)了，我要去巡山')s.connect(animal)if "__main__" == __name__: res = s.has_receivers_for(animal) print(res) res = q.has_receivers_for(animal) print(res)

TrueFalse3 基于 blinker 得 Flask 信號(hào)

Flask 集成 blinker 作為解耦應(yīng)用得解決方案。在 Flask 中，信號(hào)得使用場(chǎng)景如：請(qǐng)求到來(lái)之前，請(qǐng)求結(jié)束之后。同時(shí) Flask 也支持自定義信號(hào)。

from flask import Flaskapp = Flask(__name__)等app.route('/',methods=['GET','POST'],endpoint='index')def index(): return 'hello blinker'if __name__ == '__main__': app.run()

訪問(wèn)127.0.0.1:5000時(shí)，返回給瀏覽器hello blinker。

因?yàn)?Flask 集成了信號(hào)，所以在 Flask 中使用信號(hào)時(shí)從 Flask 中引入。

from flask import Flaskfrom flask.signals import _signalsapp = Flask(__name__)s = _signals.singal('msg')def (args): print('you have msg from ')s.connect()等app.route('/',methods=['GET','POST'],endpoint='index')def index(): s.send() return 'hello blinker'if __name__ == '__main__': app.run()3.3 Flask自帶信號(hào)

在 Flask 中除了可以自定義信號(hào)，還可以使用自帶信號(hào)。Flask 中自帶得信號(hào)有很多種，具體如下：

請(qǐng)求request_started = _signals.signal('request-started') # 請(qǐng)求到來(lái)前執(zhí)行request_finished = _signals.signal('request-finished') # 請(qǐng)求結(jié)束后執(zhí)行 模板渲染before_render_template = _signals.signal('before-render-template') # 模板渲染前執(zhí)行template_rendered = _signals.signal('template-rendered') # 模板渲染后執(zhí)行 請(qǐng)求執(zhí)行g(shù)ot_request_exception = _signals.signal('got-request-exception') # 請(qǐng)求執(zhí)行出現(xiàn)異常時(shí)執(zhí)行request_tearing_down = _signals.signal('request-tearing-down') # 請(qǐng)求執(zhí)行完畢后自動(dòng)執(zhí)行（無(wú)論成功與否）appcontext_tearing_down = _signals.signal('appcontext-tearing-down') # 請(qǐng)求上下文執(zhí)行完畢后自動(dòng)執(zhí)行（無(wú)論成功與否） 請(qǐng)求上下文中appcontext_pushed = _signals.signal('appcontext-pushed') # 請(qǐng)求上下文push時(shí)執(zhí)行appcontext_popped = _signals.signal('appcontext-popped') # 請(qǐng)求上下文pop時(shí)執(zhí)行message_flashed = _signals.signal('message-flashed') # 調(diào)用flask在其中添加數(shù)據(jù)時(shí)，自動(dòng)觸發(fā)

下面以請(qǐng)求到來(lái)之前為例，看 Flask 中信號(hào)如何使用

from flask import Flaskfrom flask.signals import _signals, request_startedimport timeapp = Flask(__name__)def wechat(args): print('you have msg from wechat')# 從flask中引入已經(jīng)定好得信號(hào)，注冊(cè)一個(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)請(qǐng)求到來(lái)時(shí)，F(xiàn)lask 會(huì)經(jīng)過(guò)request_started 通知接受方，就是函數(shù)wechat，這時(shí)wechat函數(shù)先執(zhí)行，然后才返回結(jié)果給瀏覽器。

但這種使用方法并不是很地道，因?yàn)樾盘?hào)并不支持異步方法，所以通常在生產(chǎn)環(huán)境中信號(hào)得接收者都是配置異步執(zhí)行得框架，如 Python 中大名鼎鼎得異步框架 celery。

信號(hào)得優(yōu)點(diǎn)：

1. 解耦應(yīng)用：將串行運(yùn)行得耦合應(yīng)用分解為多級(jí)執(zhí)行
2. 發(fā)布訂閱者：減少調(diào)用者得使用，一次調(diào)用通知多個(gè)訂閱者

信號(hào)得缺點(diǎn)：

1. 不支持異步
2. 支持訂閱主題得能力有限