作者：金色旭光

來(lái)源：https://www.cnblogs.com/goldsunshine/p/15426970.html

1 信號(hào)

信號(hào)是一種通知或者說(shuō)通信的方式，信號(hào)分為發(fā)送方和接收方。發(fā)送方發(fā)送一種信號(hào)，接收方收到信號(hào)的進(jìn)程會(huì)跳入信號(hào)處理函數(shù)，執(zhí)行完后再跳回原來(lái)的位置繼續(xù)執(zhí)行。

常見(jiàn)的 Linux 中的信號(hào)，通過(guò)鍵盤(pán)輸入 Ctrl+C，就是發(fā)送給系統(tǒng)一個(gè)信號(hào)，告訴系統(tǒng)退出當(dāng)前進(jìn)程。

信號(hào)的特點(diǎn)就是發(fā)送端通知訂閱者發(fā)生了什么。使用信號(hào)分為 3 步：定義信號(hào)，監(jiān)聽(tīng)信號(hào)，發(fā)送信號(hào)。

Python 中提供了信號(hào)概念的通信模塊，就是blinker。

Blinker 是一個(gè)基于 Python 的強(qiáng)大的信號(hào)庫(kù)，它既支持簡(jiǎn)單的點(diǎn)對(duì)點(diǎn)通信，也支持點(diǎn)對(duì)多點(diǎn)的組播。Flask 的信號(hào)機(jī)制就是基于它建立的。Blinker 的內(nèi)核雖然小巧，但是功能卻非常強(qiáng)大，它支持以下特性：

• 支持注冊(cè)全局命名信號(hào)

• 支持匿名信號(hào)

• 支持自定義命名信號(hào)

• 支持與接收者之間的持久連接與短暫連接

• 通過(guò)弱引用實(shí)現(xiàn)與接收者之間的自動(dòng)斷開(kāi)連接

• 支持發(fā)送任意大小的數(shù)據(jù)

• 支持收集信號(hào)接收者的返回值

• 線程安全

2 blinker 使用

安裝方法：

pip?install?blinker

2.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.connect def?animal_one(args):print(f'我是小鉆風(fēng)，今天的口號(hào)是:?{args}')@s.connect def?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.receiversprint(res)if?res:s.send()res?=?q.receiversprint(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)True False

3 基于 blinker 的 Flask 信號(hào)

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

3.1 簡(jiǎn)單 Flask demo

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。

3.2 自定義信號(hào)

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

from?flask?import?Flask from?flask.signals?import?_signalsapp?=?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自帶信號(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í)行 got_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?Flask from?flask.signals?import?_signals,?request_started import?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í)，Flask 會(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。

4 總結(jié)

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

解耦應(yīng)用：將串行運(yùn)行的耦合應(yīng)用分解為多級(jí)執(zhí)行

發(fā)布訂閱者：減少調(diào)用者的使用，一次調(diào)用通知多個(gè)訂閱者

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

不支持異步

支持訂閱主題的能力有限

各位伙伴們好，詹帥本帥搭建了一個(gè)個(gè)人博客和小程序，匯集各種干貨和資源，也方便大家閱讀，感興趣的小伙伴請(qǐng)移步小程序體驗(yàn)一下哦！（歡迎提建議）

推薦閱讀

牛逼！Python常用數(shù)據(jù)類型的基本操作（長(zhǎng)文系列第①篇）

牛逼！Python的判斷、循環(huán)和各種表達(dá)式（長(zhǎng)文系列第②篇）

牛逼！Python函數(shù)和文件操作（長(zhǎng)文系列第③篇）

牛逼！Python錯(cuò)誤、異常和模塊（長(zhǎng)文系列第④篇）

總結(jié)

以上是生活随笔為你收集整理的Python 强大的信号库 blinker 入门教程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。