大家好，我是老表～
訊號是一種通知或說通訊的方式，訊號分為發送方和接收方。發送方發送一種訊號，接收方收到訊號的進程會跳入訊號處理函數，執行完後再跳回原來的位置繼續執行。
常見的 Linux 中的訊號，透過鍵盤輸入 Ctrl C，就是傳送給系統一個訊號，告訴系統退出目前進程。
訊號的特點就是發送端通知訂閱者發生了什麼事。使用訊號分為 3 步驟：定義訊號，監聽訊號，發送訊號。
Python 中提供了訊號概念的通訊模組，就是blinker。
Blinker 是一個基於 Python 的強大的訊號庫，它既支援簡單的點對點通信，也支援點對多點的組播。 Flask 的訊號機制就是基於它所建立的。 Blinker 的核心雖然小巧，但是功能卻非常強大，它支援以下特性：
安裝方法：
pip install blinker
from blinker import signal
# 定义一个信号
s = signal('king')
def animal(args):
print('我是小钻风，大王回来了，我要去巡山')
# 信号注册一个接收者
s.connect(animal)
if "__main__" == __name__:
# 发送信号
s.send()
blinker 也支援匿名訊號，就是不需要指定一個特定的訊號值。創建的每一個匿名訊號都是互相獨立的。
from blinker import Signal
s = Signal()
def animal(sender):
print('我是小钻风，大王回来了，我要去巡山')
s.connect(animal)
if "__main__" == __name__:
s.send()
組播訊號是比較能體現出訊號優點的特性。多個接收者註冊到訊號上，發送者只需要發送一次就能傳遞訊息到多個接收者。
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('大王叫我来巡山，抓个和尚做晚餐！')
接受方支援訂閱指定的主題，只有當指定的主題傳送訊息時才傳送給接收方。這種方法很好的區分了不同的主題。
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)
除了可以函數註冊之外還有更簡單的訊號註冊方法，那就是裝飾器。
from blinker import signal
s = 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('大王叫我来巡山，抓个和尚做晚餐！')
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)
如果對於一個發送者發送訊息前要準備的耗時很長，為了避免沒有接收者導致浪費效能的情況，所以可以先檢查某一個訊號是否有接收者，在確定有接收者的情況下才會發送，做到精確。
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
print(res)
if res:
q.send()
else:
print("孩儿们都出去巡山了")
{4511880240: <weakref at 0x10d02ae80; to 'function' at 0x10cedd430 (animal)>}
我是小钻风，大王回来了，我要去巡山
{}
孩儿们都出去巡山了
也可以檢查訂閱者是否由某一個訊號
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)
print(res)
True False
Flask 整合blinker 作為解耦應用的解決方案。在 Flask 中，訊號的使用情境如：請求到來之前，請求結束之後。同時 Flask 也支援自訂訊號。
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 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 中自帶的訊號有很多種，具體如下：
请求
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函數先執行，然後才回傳結果給瀏覽器。
但這種使用方法並不是很地道，因為訊號並不支援非同步方法，所以通常在生產環境中訊號的接收者都是配置非同步執行的框架，如 Python 中大名鼎鼎的非同步框架 celery。
訊號的優點：
訊號的缺點：
#
以上是Python 強大的訊號庫：blinker的詳細內容。更多資訊請關注PHP中文網其他相關文章！
