感謝分享：金色旭光
近日：感謝分享特別cnblogs感謝原創分享者/goldsunshine/p/15426970.html
1 信號信號是一種通知或者說通信得方式，信號分為發送方和接收方。發送方發送一種信號，接收方收到信號得進程會跳入信號處理函數，執行完后再跳回原來得位置繼續執行。
常見得 Linux 中得信號，通過鍵盤輸入 Ctrl+C，就是發送給系統一個信號，告訴系統退出當前進程。
信號得特點就是發送端通知訂閱者發生了什么。使用信號分為 3 步：定義信號，監聽信號，發送信號。
Python 中提供了信號概念得通信模塊，就是blinker。
Blinker 是一個基于 Python 得強大得信號庫，它既支持簡單得點對點通信，也支持點對多點得組播。Flask 得信號機制就是基于它建立得。Blinker 得內核雖然小巧，但是功能卻非常強大，它支持以下特性：
安裝方法：
pip install blinker2.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 Signals = Signal()def animal(sender): print('我是小鉆風，大王回來了，我要去巡山')s.connect(animal)if "__main__" == __name__: s.send()2.3 組播信號
組播信號是比較能體現出信號優點得特征。多個接收者注冊到信號上，發送者只需要發送一次就能傳遞信息到多個接收者。
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 接收方訂閱主題
接受方支持訂閱指定得主題，只有當指定得主題發送消息時才發送給接收方。這種方法很好得區分了不同得主題。
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 裝飾器用法
除了可以函數注冊之外還有更簡單得信號注冊方法，那就是裝飾器。
from blinker import signals = signal('king')等s.connectdef animal_one(args): print(f'我是小鉆風，今天得口號是: {args}')等s.connectdef 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 檢查信號是否有接收者
如果對于一個發送者發送消息前要準備得耗時很長，為了避免沒有接收者導致浪費性能得情況，所以可以先檢查某一個信號是否有接收者，在確定有接收者得情況下才發送，做到精確。
from blinker import signals = 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 print(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)
TrueFalse3 基于 blinker 得 Flask 信號
Flask 集成 blinker 作為解耦應用得解決方案。在 Flask 中，信號得使用場景如：請求到來之前，請求結束之后。同時 Flask 也支持自定義信號。
3.1 簡單 Flask demofrom 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 Flaskfrom flask.signals import _signalsapp = Flask(__name__)s = _signals.singal('msg')def 感謝對創作者的支持(args): print('you have msg from 感謝對創作者的支持')s.connect(感謝對創作者的支持)等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 Flaskfrom flask.signals import _signals, request_startedimport timeapp = 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函數先執行，然后才返回結果給瀏覽器。
但這種使用方法并不是很地道，因為信號并不支持異步方法，所以通常在生產環境中信號得接收者都是配置異步執行得框架，如 Python 中大名鼎鼎得異步框架 celery。
4 總結信號得優點：
- 解耦應用：將串行運行得耦合應用分解為多級執行
- 發布訂閱者：減少調用者得使用，一次調用通知多個訂閱者
信號得缺點：
- 不支持異步
- 支持訂閱主題得能力有限
