python中的信號(hào)通信 blinker的使用小結(jié)
信號(hào):
信號(hào)是一種通知或者說通信的方式,信號(hào)分為發(fā)送方和接收方。發(fā)送方發(fā)送一中信號(hào),接收方收到信號(hào)的進(jìn)程會(huì)跳入信號(hào)處理函數(shù),執(zhí)行完后再跳回原來的位置繼續(xù)執(zhí)行。常見的linux中的信號(hào),通過鍵盤輸入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)聽信號(hào),發(fā)送信號(hào)
python中提供了信號(hào)概念的通信模塊,就是blinker
。
官方介紹:
Blinker 是一個(gè)基于Python的強(qiáng)大的信號(hào)庫,它既支持簡單的點(diǎn)對(duì)點(diǎn)通信,也支持點(diǎn)對(duì)多點(diǎn)的組播。Flask的信號(hào)機(jī)制就是基于它建立的。Blinker的內(nèi)核雖然小巧,但是功能卻非常強(qiáng)大,它支持以下特性:
- 支持注冊全局命名信號(hào)
- 支持匿名信號(hào)
- 支持自定義命名信號(hào)
- 支持與接收者之間的持久連接與短暫連接
- 通過弱引用實(shí)現(xiàn)與接收者之間的自動(dòng)斷開連接
- 支持發(fā)送任意大小的數(shù)據(jù)
- 支持收集信號(hào)接收者的返回值
- 線程安全
blinker 使用
安裝方法:
pip install blinker
命名信號(hào)
from blinker import signal # 定義一個(gè)信號(hào) s = signal('king') def animal(args): print('我是小鉆風(fēng),大王回來了,我要去巡山') # 信號(hào)注冊一個(gè)接收者 s.connect(animal) if "__main__" == __name__: # 發(fā)送信號(hào) s.send()
匿名信號(hào)
blinker也支持匿名信號(hào),就是不需要指定一個(gè)具體的信號(hào)值。創(chuàng)建的每一個(gè)匿名信號(hào)都是互相獨(dú)立的。
from blinker import Signal s = Signal() def animal(sender): print('我是小鉆風(fēng),大王回來了,我要去巡山') s.connect(animal) if "__main__" == __name__: s.send()
組播信號(hào)
組播信號(hào)是比較能體現(xiàn)出信號(hào)優(yōu)點(diǎn)的特征。多個(gè)接收者注冊到信號(hào)上,發(fā)送者只需要發(fā)送一次就能傳遞信息到多個(gè)接收者。
from blinker import signal s = 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('大王叫我來巡山,抓個(gè)和尚做晚餐!')
接收方訂閱主題
接受方支持訂閱指定的主題,只有當(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)
裝飾器用法
除了可以函數(shù)注冊之外還有更簡單的信號(hào)注冊方法,那就是裝飾器。
from blinker import signal s = 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('大王叫我來巡山,抓個(gè)和尚做晚餐!')
可訂閱主題的裝飾器
connect
的注冊方法用著裝飾器時(shí)有一個(gè)弊端就是不能夠訂閱主題,所以有更高級(jí)的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)
檢查信號(hào)是否有接收者
如果對(duì)于一個(gè)發(fā)送者發(fā)送消息前要準(zhǔn)備的耗時(shí)很長,為了避免沒有接收者導(dǎo)致浪費(fèi)性能的情況,所以可以先檢查某一個(gè)信號(hào)是否有接收者,在確定有接收者的情況下才發(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: <weakref at 0x10d02ae80; to 'function' at 0x10cedd430 (animal)>} 我是小鉆風(fēng),大王回來了,我要去巡山 {} 孩兒們都出去巡山了
檢查訂閱者是否訂閱了某個(gè)信號(hào)
也可以檢查訂閱者是否由某一個(gè)信號(hào)
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
基于blinker的Flask信號(hào)
Flask集成blinker作為解耦應(yīng)用的解決方案。在Flask中,信號(hào)的使用場景如:請(qǐng)求到來之前,請(qǐng)求結(jié)束之后。同時(shí)Flask也支持自定義信號(hào)。
簡單 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
。
自定義信號(hào)
因?yàn)閒lask集成了信號(hào),所以在flask中使用信號(hào)時(shí)從flask中引入。
from flask.signals import _signals
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()
Flask自帶信號(hào)
在Flask中除了可以自定義信號(hào),還可以使用自帶信號(hào)。Flask中自帶的信號(hào)有很多種,具體如下:
請(qǐng)求 request_started = _signals.signal('request-started') # 請(qǐng)求到來前執(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í)行(無論成功與否) appcontext_tearing_down = _signals.signal('appcontext-tearing-down') # 請(qǐng)求上下文執(zhí)行完畢后自動(dòng)執(zhí)行(無論成功與否) 請(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)求到來之前為例,看flask中信號(hào)如何使用
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)定好的信號(hào),注冊一個(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)求到來時(shí),flask會(huì)經(jīng)過request_started
通知接受方,就是函數(shù)wechat
,這時(shí)wechat
函數(shù)先執(zhí)行,然后才返回結(jié)果給瀏覽器。
但這種使用方法并不是很地道,因?yàn)樾盘?hào)并不支持異步方法,所以通常在生產(chǎn)環(huán)境中信號(hào)的接收者都是配置異步執(zhí)行的框架,如python中大名鼎鼎的異步框架celery。
總結(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):
- 不支持異步
- 支持訂閱主題的能力有限
到此這篇關(guān)于python中的信號(hào)通信 blinker的文章就介紹到這了,更多相關(guān)python信號(hào)blinker內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
python實(shí)現(xiàn)類似ftp傳輸文件的網(wǎng)絡(luò)程序示例
這篇文章主要介紹了python實(shí)現(xiàn)類似ftp傳輸文件的網(wǎng)絡(luò)程序示例,需要的朋友可以參考下2014-04-04Jupyter Notebook讀入csv文件時(shí)出錯(cuò)的解決方案
這篇文章主要介紹了Jupyter Notebook讀入csv文件時(shí)出錯(cuò)的解決方案,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2021-03-03Python爬蟲框架scrapy實(shí)現(xiàn)downloader_middleware設(shè)置proxy代理功能示例
這篇文章主要介紹了Python爬蟲框架scrapy實(shí)現(xiàn)downloader_middleware設(shè)置proxy代理功能,結(jié)合實(shí)例形式分析了scrapy框架proxy代理設(shè)置技巧與相關(guān)問題注意事項(xiàng),需要的朋友可以參考下2018-08-08python cs架構(gòu)實(shí)現(xiàn)簡單文件傳輸
這篇文章主要為大家詳細(xì)介紹了python cs架構(gòu)實(shí)現(xiàn)簡單文件傳輸,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2018-07-07Python cookbook(數(shù)據(jù)結(jié)構(gòu)與算法)通過公共鍵對(duì)字典列表排序算法示例
這篇文章主要介紹了Python cookbook(數(shù)據(jù)結(jié)構(gòu)與算法)通過公共鍵對(duì)字典列表排序算法,結(jié)合實(shí)例形式分析了Python基于operator模塊中的itemgetter()函數(shù)對(duì)字典進(jìn)行排序的相關(guān)操作技巧,需要的朋友可以參考下2018-03-03利用Python實(shí)現(xiàn)自動(dòng)工作匯報(bào)的腳本分享
這篇文章主要為大家詳細(xì)介紹了如何利用Python實(shí)現(xiàn)一個(gè)自動(dòng)工作匯報(bào)的腳本,文中的示例代碼講解詳細(xì),對(duì)我們學(xué)習(xí)Python有一定幫助,需要的可以參考一下2022-08-08JupyterNotebook設(shè)置Python環(huán)境的方法步驟
這篇文章主要介紹了JupyterNotebook設(shè)置Python環(huán)境的方法步驟,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-12-12python實(shí)現(xiàn)bitmap數(shù)據(jù)結(jié)構(gòu)詳解
bitmap是很常用的數(shù)據(jù)結(jié)構(gòu),比如用于Bloom Filter中,下面是使用python實(shí)現(xiàn)bitmap數(shù)據(jù)結(jié)構(gòu)的代碼講解,需要的朋友可以參考下2014-02-02