欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

python中的信號通信 blinker的使用小結(jié)

 更新時間:2021年10月22日 08:58:57   作者:金色旭光  
信號是一種通知或者說通信的方式,信號分為發(fā)送方和接收方,信號的特點就是發(fā)送端通知訂閱者發(fā)生了什么,今天通過本文給大家介紹python中的信號通信 blinker的相關(guān)知識,感興趣的朋友一起看看吧

信號:

信號是一種通知或者說通信的方式,信號分為發(fā)送方和接收方。發(fā)送方發(fā)送一中信號,接收方收到信號的進程會跳入信號處理函數(shù),執(zhí)行完后再跳回原來的位置繼續(xù)執(zhí)行。常見的linux中的信號,通過鍵盤輸入Ctrl+C,就是發(fā)送給系統(tǒng)一個信號,告訴系統(tǒng)退出當(dāng)前進程。

信號的特點就是發(fā)送端通知訂閱者發(fā)生了什么。使用信號分為3步,定義信號,監(jiān)聽信號,發(fā)送信號

python中提供了信號概念的通信模塊,就是blinker。

官方介紹:

Blinker 是一個基于Python的強大的信號庫,它既支持簡單的點對點通信,也支持點對多點的組播。Flask的信號機制就是基于它建立的。Blinker的內(nèi)核雖然小巧,但是功能卻非常強大,它支持以下特性:

  • 支持注冊全局命名信號
  • 支持匿名信號
  • 支持自定義命名信號
  • 支持與接收者之間的持久連接與短暫連接
  • 通過弱引用實現(xiàn)與接收者之間的自動斷開連接
  • 支持發(fā)送任意大小的數(shù)據(jù)
  • 支持收集信號接收者的返回值
  • 線程安全

blinker 使用

安裝方法:

pip install blinker

命名信號

from blinker import signal

# 定義一個信號
s = signal('king')


def animal(args):
    print('我是小鉆風(fēng),大王回來了,我要去巡山')

# 信號注冊一個接收者
s.connect(animal)

if "__main__" == __name__:
    # 發(fā)送信號
    s.send()

匿名信號

blinker也支持匿名信號,就是不需要指定一個具體的信號值。創(chuàng)建的每一個匿名信號都是互相獨立的。

from blinker import Signal

s = Signal()

def animal(sender):
    print('我是小鉆風(fēng),大王回來了,我要去巡山')

s.connect(animal)

if "__main__" == __name__:
    s.send()

組播信號

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

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('大王叫我來巡山,抓個和尚做晚餐!')

接收方訂閱主題

接受方支持訂閱指定的主題,只有當(dāng)指定的主題發(fā)送消息時才發(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ù)注冊之外還有更簡單的信號注冊方法,那就是裝飾器。

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('大王叫我來巡山,抓個和尚做晚餐!')

可訂閱主題的裝飾器

connect的注冊方法用著裝飾器時有一個弊端就是不能夠訂閱主題,所以有更高級的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)

檢查信號是否有接收者

如果對于一個發(fā)送者發(fā)送消息前要準(zhǔn)備的耗時很長,為了避免沒有接收者導(dǎ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),大王回來了,我要去巡山
{}
孩兒們都出去巡山了

檢查訂閱者是否訂閱了某個信號

也可以檢查訂閱者是否由某一個信號

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信號

Flask集成blinker作為解耦應(yīng)用的解決方案。在Flask中,信號的使用場景如:請求到來之前,請求結(jié)束之后。同時Flask也支持自定義信號。

簡單 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時,返回給瀏覽器hello blinker

自定義信號

因為flask集成了信號,所以在flask中使用信號時從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自帶信號

在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 time

app = 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()

當(dāng)請求到來時,flask會經(jīng)過request_started 通知接受方,就是函數(shù)wechat,這時wechat函數(shù)先執(zhí)行,然后才返回結(jié)果給瀏覽器。

但這種使用方法并不是很地道,因為信號并不支持異步方法,所以通常在生產(chǎn)環(huán)境中信號的接收者都是配置異步執(zhí)行的框架,如python中大名鼎鼎的異步框架celery。

總結(jié)

信號的優(yōu)點:

  • 解耦應(yīng)用:將串行運行的耦合應(yīng)用分解為多級執(zhí)行
  • 發(fā)布訂閱者:減少調(diào)用者的使用,一次調(diào)用通知多個訂閱者

信號的缺點:

  • 不支持異步
  • 支持訂閱主題的能力有限

到此這篇關(guān)于python中的信號通信 blinker的文章就介紹到這了,更多相關(guān)python信號blinker內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

您可能感興趣的文章:

相關(guān)文章

最新評論