Python 強大的訊號庫 blinker 入門詳細教學

2022-02-19 16:00:51

1、訊號

訊號是一種通知或者說通訊的方式，訊號分為傳送方和接收方。傳送方傳送一種訊號，接收方收到訊號的程序會跳入訊號處理常式，執行完後再跳回原來的位置繼續執行。

常見的 Linux 中的訊號，通過鍵盤輸入 Ctrl+C，就是傳送給系統一個訊號，告訴系統退出當前程序。

訊號的特點就是傳送端通知訂閱者發生了什麼。使用訊號分為 3 步：定義訊號，監聽訊號，傳送訊號。

Python 中提供了訊號概念的通訊模組，就是blinker。

Blinker 是一個基於 Python 的強大的訊號庫，它既支援簡單的對等通訊，也支援點對多點的組播。Flask 的訊號機制就是基於它建立的。Blinker 的核心雖然小巧，但是功能卻非常強大，它支援以下特性：

支援註冊全域性命名訊號
支援匿名訊號
支援自定義命名訊號
支援與接收者之間的持久連線與短暫連線
通過弱參照實現與接收者之間的自動斷開連線
支援傳送任意大小的資料
支援收集訊號接收者的返回值
執行緒安全

2、blinker 使用

安裝方法：

pip install blinker

2.1 命名訊號

from blinker import signal
 
# 定義一個訊號
s = signal('king')
def animal(args):
    print('我是小鑽風，大王回來了，我要去巡山')
# 訊號註冊一個接收者
s.connect(animal)
if "__main__" == __name__:
    # 傳送訊號
    s.send()

2.2 匿名訊號

blinker 也支援匿名訊號，就是不需要指定一個具體的訊號值。建立的每一個匿名訊號都是互相獨立的。

from blinker import Signal
 
s = Signal()
def animal(sender):
    print('我是小鑽風，大王回來了，我要去巡山')
s.connect(animal)
if "__main__" == __name__:
    s.send()

2.3 組播訊號

組播訊號是比較能體現出訊號優點的特徵。多個接收者註冊到訊號上，傳送者只需要傳送一次就能傳遞資訊到多個接收者。

from blinker import signal
 
s = 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 接收方訂閱主題

接受方支援訂閱指定的主題，只有當指定的主題傳送訊息時才傳送給接收方。這種方法很好的區分了不同的主題。

from blinker import signal
 
s = signal('king')
def animal(args):
    print(f'我是小鑽風，{args} 是我大哥')
s.connect(animal, sender='大象')
if "__main__" == __name__:
    for i in ['獅子', '大象', '大鵬']:
        s.send(i)

2.5 裝飾器用法

除了可以函數註冊之外還有更簡單的訊號註冊方法，那就是裝飾器。

from blinker import signal
 
s = signal('king')
@s.connect
def animal_one(args):
    print(f'我是小鑽風，今天的口號是: {args}')
def animal_two(args):
    print(f'我是大鑽風，今天的口號是: {args}')
if "__main__" == __name__:
    s.send('大王叫我來巡山，抓個和尚做晚餐！')

2.6 可訂閱主題的裝飾器

connect的註冊方法用著裝飾器時有一個弊端就是不能夠訂閱主題，所以有更高階的connect_via方法支援訂閱主題。

from blinker import signal
s = signal('king')
@s.connect_via('大象')
def animal(args):
    print(f'我是小鑽風，{args} 是我大哥')
if "__main__" == __name__:
    for i in ['獅子', '大象', '大鵬']:
        s.send(i)

2.7 檢查訊號是否有接收者

如果對於一個傳送者傳送訊息前要準備的耗時很長，為了避免沒有接收者導致浪費效能的情況，所以可以先檢查某一個訊號是否有接收者，在確定有接收者的情況下才傳送，做到精確。

from blinker import signal
s = signal('king')
q = signal('queue')
def animal(sender):
    print('我是小鑽風，大王回來了，我要去巡山')
s.connect(animal)
if "__main__" == __name__:
    
    res = s.receivers
    print(res)
    if res:
        s.send()
    res = q.receivers
        q.send()
    else:
        print("孩兒們都出去巡山了")

{4511880240: <weakref at 0x10d02ae80; to 'function' at 0x10cedd430 (animal)>}
我是小鑽風，大王回來了，我要去巡山
{}
孩兒們都出去巡山了

2.8 檢查訂閱者是否訂閱了某個訊號

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

from blinker import signal
s = 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)

True
False

3、基於 blinker 的 Flask 訊號

Flask 整合 blinker 作為解耦應用的解決方案。在 Flask 中，訊號的使用場景如：請求到來之前，請求結束之後。同時 Flask 也支援自定義訊號。

3.1 簡單 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。

3.2 自定義訊號

因為 Flask 整合了訊號，所以在 Flask 中使用訊號時從 Flask 中引入。

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

3.3 Flask自帶訊號

在 Flask 中除了可以自定義訊號，還可以使用自帶訊號。Flask 中自帶的訊號有很多種，具體如下：

請求
request_started = _signals.signal('request-started')                # 請求到來前執行
request_finished = _signals.signal('request-finished')              # 請求結束後執行
 
模板渲染
before_render_template = _signals.signal('before-render-template')  # 模板渲染前執行
template_rendered = _signals.signal('template-rendered')            # 模板渲染後執行
請求執行
got_request_exception = _signals.signal('got-request-exception')    # 請求執行出現異常時執行
request_tearing_down = _signals.signal('request-tearing-down')      # 請求執行完畢後自動執行（無論成功與否）
appcontext_tearing_down = _signals.signal('appcontext-tearing-down') # 請求上下文執行完畢後自動執行（無論成功與否）
請求上下文中
appcontext_pushed = _signals.signal('appcontext-pushed')            # 請求上下文push時執行
appcontext_popped = _signals.signal('appcontext-popped')            # 請求上下文pop時執行
message_flashed = _signals.signal('message-flashed')                # 呼叫flask在其中新增資料時，自動觸

下面以請求到來之前為例，看 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中引入已經定好的訊號，註冊一個函數
request_started.connect(wechat)
@app.route('/',methods=['GET','POST'],endpoint='index')
def index():
    return 'hello blinker'
if __name__ == '__main__':
    app.run()

當請求到來時，Flask 會經過request_started 通知接受方，就是函數wechat，這時wechat函數先執行，然後才返回結果給瀏覽器。