Python's powerful signal library: blinker

1. Signal

Hello everyone, I am my old cousin~

A signal is a way of notification or communication. Signals are divided into sending party and receiver. The sender sends a signal, and the process of the receiver receiving the signal will jump into the signal processing function. After execution, it will jump back to the original position to continue execution.

Common signals in Linux, inputting Ctrl C through the keyboard is to send a signal to the system to tell the system to exit the current process.

The characteristic of a signal is that the sender notifies the subscriber of what happened. Using signals is divided into 3 steps: defining signals, monitoring signals, and sending signals.

#The communication module that provides the signal concept in Python is blinker.

Blinker is a powerful signal library based on Python, which supports both simple point-to-point communication and point-to-multipoint multicast. Flask's signaling mechanism is based on it. Although the kernel of Blinker is small, it is very powerful. It supports the following features:

• Supports registration of global named signals
• Supports anonymous signals
• Supports custom naming Signal
• Supports persistent connections and short-lived connections with the receiver
• Implements automatic disconnection with the receiver through weak references
• Supports sending any size Data
• Supports collecting the return value of the signal receiver
• Thread safety

2. Blinker uses the

installation method:

pip install blinker

2.1 Named signal

from blinker import signal
# 定义一个信号
s = signal('king')
def animal(args):
print('我是小钻风，大王回来了，我要去巡山')
# 信号注册一个接收者
s.connect(animal)
if "__main__" == __name__:
# 发送信号
s.send()

2.2 Anonymous signal

blinker also supports anonymous signals, that is, there is no need to specify a specific signal value. Each anonymous signal created is independent of each other.

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

2.3 Multicast signal

Multicast signal is a characteristic that can better reflect the advantages of signals. Multiple receivers are registered to the signal, and the sender only needs to send once to deliver information to multiple receivers.

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 The receiver subscribes to the topic

The receiver supports subscribing to the specified topic, and only when the specified topic sends a message, it is sent to the receiver. This method distinguishes different topics well.

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 Decorator Usage

In addition to function registration, there is a simpler signal registration method, that is, decorator.

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('大王叫我来巡山，抓个和尚做晚餐！')

2.6 Decorators that can subscribe to topics

Connect's registration method has a drawback when using decorators, which is that it cannot subscribe to topics, so there is a more advanced connect_via method that supports subscribing to topics.

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 Check whether there is a receiver for the signal

If it takes a long time for a sender to prepare before sending a message, in order to avoid the waste of performance caused by no receiver, you can first Check whether there is a receiver for a certain signal, and only send it when it is confirmed that there is a receiver, so as to be accurate.

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)>}
我是小钻风，大王回来了，我要去巡山
{}
孩儿们都出去巡山了

2.8 Check whether the subscriber is subscribed to a certain signal

You can also check whether the subscriber is subscribed to a certain signal

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

3. Flask signal based on blinker

Flask integrates blinker as a solution for decoupled applications. In Flask, the usage scenarios of signals are as follows: before the request arrives and after the request ends. At the same time, Flask also supports custom signals.

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

When accessing 127.0.0.1:5000, return hello blinker to the browser.

3.2 Custom signals

Because Flask integrates signals, they are imported from Flask when using signals in 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’s own signals

In addition to customizing signals, you can also use built-in signals in Flask. There are many kinds of signals that come with Flask, as follows:

请求
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在其中添加数据时，自动触发

The following takes before the request arrives as an example to see how to use signals in 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()

When a request arrives, Flask will go through request_started Notifying the recipient is the function wechat. At this time, the wechat function is executed first and then returns the result to the browser.

But this method of use is not very authentic, because signals do not support asynchronous methods, so usually the receivers of signals in production environments are frameworks configured for asynchronous execution, such as the famous asynchronous framework celery in Python.

4. Summary

Advantages of signals:

1. Decoupled applications: Decompose serially running coupled applications into multi-level execution
2. Publish Subscribers: Reduce the use of callers, notify multiple subscribers with one call

Disadvantages of signals:

1. Does not support asynchronous
2. Supports subscriptions The theme's capacity is limited

Every like, view, message, and retweet of yours is very important to me, thank you for your support.

Okay, see you next time. I love cats, technology, and Sisi’s old cousin even more⁽⁽ଘ( ˙꒳˙ )ଓ⁾⁾

The above is the detailed content of Python's powerful signal library: blinker. For more information, please follow other related articles on the PHP Chinese website!