首页　>　文章列表　>　Python 强大的信号库：blinker

Python 强大的信号库：blinker

Python blinker 信号库
376 2023-04-24

一. 信号

大家好，我是老表～

信号是一种通知或者说通信的方式，信号分为发送方和接收方。发送方发送一种信号，接收方收到信号的进程会跳入信号处理函数，执行完后再跳回原来的位置继续执行。

常见的 Linux 中的信号，通过键盘输入 Ctrl+C，就是发送给系统一个信号，告诉系统退出当前进程。

信号的特点就是发送端通知订阅者发生了什么。使用信号分为 3 步：定义信号，监听信号，发送信号。

Python 中提供了信号概念的通信模块，就是blinker。

Blinker 是一个基于 Python 的强大的信号库，它既支持简单的点对点通信，也支持点对多点的组播。Flask 的信号机制就是基于它建立的。Blinker 的内核虽然小巧，但是功能却非常强大，它支持以下特性：

支持注册全局命名信号
支持匿名信号
支持自定义命名信号
支持与接收者之间的持久连接与短暂连接
通过弱引用实现与接收者之间的自动断开连接
支持发送任意大小的数据
支持收集信号接收者的返回值
线程安全

二. blinker 使用

安装方法：

2.5 装饰器用法除了可以函数注册之外还有更简单的信号注册方法，那就是装饰器。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函数先执行，然后才返回结果给浏览器。