Python 强大的信号库 blinker 入门教程
作者:金色旭光
來源:https://www.cnblogs.com/goldsunshine/p/15426970.html
1 信號
信號是一種通知或者說通信的方式,信號分為發(fā)送方和接收方。發(fā)送方發(fā)送一種信號,接收方收到信號的進程會跳入信號處理函數(shù),執(zhí)行完后再跳回原來的位置繼續(xù)執(zhí)行。
常見的 Linux 中的信號,通過鍵盤輸入 Ctrl+C,就是發(fā)送給系統(tǒng)一個信號,告訴系統(tǒng)退出當前進程。
信號的特點就是發(fā)送端通知訂閱者發(fā)生了什么。使用信號分為 3 步:定義信號,監(jiān)聽信號,發(fā)送信號。
Python 中提供了信號概念的通信模塊,就是blinker。
Blinker 是一個基于 Python 的強大的信號庫,它既支持簡單的點對點通信,也支持點對多點的組播。Flask 的信號機制就是基于它建立的。Blinker 的內(nèi)核雖然小巧,但是功能卻非常強大,它支持以下特性:
支持注冊全局命名信號
支持匿名信號
支持自定義命名信號
支持與接收者之間的持久連接與短暫連接
通過弱引用實現(xiàn)與接收者之間的自動斷開連接
支持發(fā)送任意大小的數(shù)據(jù)
支持收集信號接收者的返回值
線程安全
2 blinker 使用
安裝方法:
pip?install?blinker2.1 命名信號
from?blinker?import?signal#?定義一個信號 s?=?signal('king')def?animal(args):print('我是小鉆風,大王回來了,我要去巡山')#?信號注冊一個接收者 s.connect(animal)if?"__main__"?==?__name__:#?發(fā)送信號s.send()2.2 匿名信號
blinker 也支持匿名信號,就是不需要指定一個具體的信號值。創(chuàng)建的每一個匿名信號都是互相獨立的。
from?blinker?import?Signals?=?Signal()def?animal(sender):print('我是小鉆風,大王回來了,我要去巡山')s.connect(animal)if?"__main__"?==?__name__:s.send()2.3 組播信號
組播信號是比較能體現(xiàn)出信號優(yōu)點的特征。多個接收者注冊到信號上,發(fā)送者只需要發(fā)送一次就能傳遞信息到多個接收者。
from?blinker?import?signals?=?signal('king')def?animal_one(args):print(f'我是小鉆風,今天的口號是:?{args}')def?animal_two(args):print(f'我是大鉆風,今天的口號是:?{args}')s.connect(animal_one) s.connect(animal_two)if?"__main__"?==?__name__:s.send('大王叫我來巡山,抓個和尚做晚餐!')2.4 接收方訂閱主題
接受方支持訂閱指定的主題,只有當指定的主題發(fā)送消息時才發(fā)送給接收方。這種方法很好的區(qū)分了不同的主題。
from?blinker?import?signals?=?signal('king')def?animal(args):print(f'我是小鉆風,{args}?是我大哥')s.connect(animal,?sender='大象')if?"__main__"?==?__name__:for?i?in?['獅子',?'大象',?'大鵬']:s.send(i)2.5 裝飾器用法
除了可以函數(shù)注冊之外還有更簡單的信號注冊方法,那就是裝飾器。
from?blinker?import?signals?=?signal('king')@s.connect def?animal_one(args):print(f'我是小鉆風,今天的口號是:?{args}')@s.connect def?animal_two(args):print(f'我是大鉆風,今天的口號是:?{args}')if?"__main__"?==?__name__:s.send('大王叫我來巡山,抓個和尚做晚餐!')2.6 可訂閱主題的裝飾器
connect的注冊方法用著裝飾器時有一個弊端就是不能夠訂閱主題,所以有更高級的connect_via方法支持訂閱主題。
from?blinker?import?signals?=?signal('king')@s.connect_via('大象') def?animal(args):print(f'我是小鉆風,{args}?是我大哥')if?"__main__"?==?__name__:for?i?in?['獅子',?'大象',?'大鵬']:s.send(i)2.7 檢查信號是否有接收者
如果對于一個發(fā)送者發(fā)送消息前要準備的耗時很長,為了避免沒有接收者導(dǎo)致浪費性能的情況,所以可以先檢查某一個信號是否有接收者,在確定有接收者的情況下才發(fā)送,做到精確。
from?blinker?import?signals?=?signal('king') q?=?signal('queue')def?animal(sender):print('我是小鉆風,大王回來了,我要去巡山')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)>} 我是小鉆風,大王回來了,我要去巡山 {} 孩兒們都出去巡山了2.8 檢查訂閱者是否訂閱了某個信號
也可以檢查訂閱者是否由某一個信號
from?blinker?import?signals?=?signal('king') q?=?signal('queue')def?animal(sender):print('我是小鉆風,大王回來了,我要去巡山')s.connect(animal)if?"__main__"?==?__name__:res?=?s.has_receivers_for(animal)print(res)res?=?q.has_receivers_for(animal)print(res)True False3 基于 blinker 的 Flask 信號
Flask 集成 blinker 作為解耦應(yīng)用的解決方案。在 Flask 中,信號的使用場景如:請求到來之前,請求結(jié)束之后。同時 Flask 也支持自定義信號。
3.1 簡單 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()訪問127.0.0.1:5000時,返回給瀏覽器hello blinker。
3.2 自定義信號
因為 Flask 集成了信號,所以在 Flask 中使用信號時從 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自帶信號
在 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)異常時執(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時執(zhí)行 appcontext_popped?=?_signals.signal('appcontext-popped')????????????#?請求上下文pop時執(zhí)行message_flashed?=?_signals.signal('message-flashed')????????????????#?調(diào)用flask在其中添加數(shù)據(jù)時,自動觸發(fā)下面以請求到來之前為例,看 Flask 中信號如何使用
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)定好的信號,注冊一個函數(shù) request_started.connect(wechat)@app.route('/',methods=['GET','POST'],endpoint='index') def?index():return?'hello?blinker'if?__name__?==?'__main__':app.run()當請求到來時,Flask 會經(jīng)過request_started 通知接受方,就是函數(shù)wechat,這時wechat函數(shù)先執(zhí)行,然后才返回結(jié)果給瀏覽器。
但這種使用方法并不是很地道,因為信號并不支持異步方法,所以通常在生產(chǎn)環(huán)境中信號的接收者都是配置異步執(zhí)行的框架,如 Python 中大名鼎鼎的異步框架 celery。
4 總結(jié)
信號的優(yōu)點:
解耦應(yīng)用:將串行運行的耦合應(yīng)用分解為多級執(zhí)行
發(fā)布訂閱者:減少調(diào)用者的使用,一次調(diào)用通知多個訂閱者
信號的缺點:
不支持異步
支持訂閱主題的能力有限
各位伙伴們好,詹帥本帥搭建了一個個人博客和小程序,匯集各種干貨和資源,也方便大家閱讀,感興趣的小伙伴請移步小程序體驗一下哦!(歡迎提建議)
推薦閱讀
牛逼!Python常用數(shù)據(jù)類型的基本操作(長文系列第①篇)
牛逼!Python的判斷、循環(huán)和各種表達式(長文系列第②篇)
牛逼!Python函數(shù)和文件操作(長文系列第③篇)
牛逼!Python錯誤、異常和模塊(長文系列第④篇)
總結(jié)
以上是生活随笔為你收集整理的Python 强大的信号库 blinker 入门教程的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: Python Bokeh 库进行数据可视
- 下一篇: 以后再也不去字节面试了…