Shulou(Shulou.com)06/03 Report--

This article mainly introduces "how to use the signal communication blinker in python". In the daily operation, I believe that many people have doubts about how to use the signal communication blinker in python. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts about "how to use the signal communication blinker in python". Next, please follow the editor to study!

Signal:

A signal is a way of notification or communication, and the signal is divided into sender and receiver. The sender sends a signal, and the process that the receiver receives the signal jumps into the signal processing function, and then jumps back to the original position to continue execution. The common signal in linux, entering 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 the signal is that the sender informs the subscriber what has happened. The use of the signal is divided into three steps: define the signal, monitor the signal, and send the signal.

The communication module of signal concept is provided in python, which is blinker.

Official introduction:

Blinker is a powerful signal library based on Python, which supports not only simple point-to-point communication, but also point-to-multipoint multicast. The signaling mechanism of Flask is based on it. Although the kernel of Blinker is small, it is very powerful. It supports the following features:

Support for registering global named signals

Anonymous signals are supported

Support for custom named signals

Support persistent and transient connections to the recipient

Automatic disconnection from the recipient through weak references

Support for sending data of any size

Support to collect the return value of the signal receiver

Thread safety

Blinker usage

Installation method:

Pip install blinker

Named signal from blinker import signal# defines a signal s = signal ('king') def animal (args): print (' I am Xiao Dengfeng, the king is back, I'm going to tour the mountains') # signal registers a receiver s.connect (animal) if "_ _ main__" = = _ _ name__: # send signal s.send ()

Anonymous signal

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

From blinker import Signals = Signal () def animal (sender): print ('I'm Xiaofeng, the king is back, I'm going to patrol the mountains') s.connect (animal) if "_ _ main__" = _ _ name__: s.send () Multicast signal

Multicast signal is a characteristic that can reflect the advantages of the signal. Multiple receivers register with the signal, and the sender only needs to send the message once to transmit the information to multiple receivers.

From blinker import signals = signal ('king') def animal_one (args): print (f' I am a small drill wind, today's slogan is: {args}') def animal_two (args): print (I am a big drill wind, today's slogan is: {args}') s.connect (animal_one) s.connect (animal_two) if "_ _ main__" = _ _ name__: s.send ('the king asked me to patrol the mountains Grab a monk for dinner!')

Recipients subscribe to topics

The recipient supports subscribing to the specified topic and sends it to the recipient only when the specified topic sends a message. This approach makes a good distinction between different themes.

From blinker import signals = signal ('king') def animal (args): print (f' I am drillwind, {args} is my eldest brother') s.connect (animal, sender=' Elephant') if "_ main__" = = _ name__: for i in ['Lion', 'Elephant', 'Dapeng']: s.send (I)

Usage of decorator

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

From blinker import signals = signal ('king') @ s.connectdef animal_one (args): print (I am a small drill wind, today's slogan is: {args}') @ s.connectdef animal_two (args): print (I am a big drill wind, today's slogan is: {args}') if "_ main__" = _ _ name__: s.send ('the king asked me to patrol the mountains and grab a monk to make dinner!') A topic-subscribing decorator

One drawback of connect's registration method using decorators is that you cannot subscribe to topics, so there are more advanced connect_via methods that support subscribing to topics.

From blinker import signals = signal ('king') @ s.connect_via (' Elephant') def animal (args): print (f'I am a drillwind, {args} is my eldest brother') if "_ _ main__" = = _ name__: for i in ['Lion', 'Elephant', 'Dapeng']: s.send (I) check whether the signal has a receiver

If it takes a long time for a sender to prepare before sending a message, in order to avoid wasting performance without a receiver, you can first check whether a signal has a receiver and send it only when it is certain that there is a receiver. Be accurate.

From blinker import signals = signal ('king') Q = signal (' queue') def animal (sender): print ('I am Xiao Dengfeng, the king is back I'm going on a mountain tour') 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 ("the kids are out on a mountain tour") {4511880240:} I'm a little drill wind, and the king is back. I'm going on a mountain tour. {} the kids are out on a mountain tour to see if the subscribers subscribe to a certain signal.

You can also check whether the subscriber is caused by a certain signal

From blinker import signals = signal ('king') Q = signal (' queue') def animal (sender): print ('I'm a little drill wind, the king is back, I'm going to tour the mountains') s.connect (animal) if "_ main__" = _ _ name__: res = s.has_receivers_for (animal) print (res) res = q.has_receivers_for (animal) print (res) TrueFalse Flask signal based on blinker

Flask integrates blinker as a solution for decoupling applications. In Flask, signal usage scenarios such as before the request arrives and after the request ends. Flask also supports custom signals.

Simple Flask demofrom flask import Flaskapp = Flask (_ _ name__) @ app.route ('/', methods= ['GET','POST'], endpoint='index') def index (): return' hello blinker'if _ _ name__ ='_ _ main__': app.run ()

When you access 127.0.0.1 hello blinker 5000, return it to the browser.

Custom signal

Because flask integrates signals, it is introduced from flask when signals are used in flask.

From flask.signals import _ signalsfrom flask import Flaskfrom flask.signals import _ signalsapp = 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 self-contained signal

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

Request request_started = _ signals.signal ('request-started') # execute request_finished = _ signals.signal (' request-finished') # perform template rendering after the request ends before_render_template = _ signals.signal ('before-render-template') # execute template_rendered = _ signals.signal (' template-rendered') before template rendering # request for execution after template rendering got_request_exception = _ signals.signal ('got-request-exception') # execute when an exception occurs in the request execution request_tearing_down = _ signals.signal (' request-tearing-down') # automatically execute after the request is completed (whether successful or not) appcontext_tearing_down = _ signals.signal ('appcontext-tearing-down') # request context execution Automatic execution after completion (whether successful or not) appcontext_pushed = _ signals.signal ('appcontext-pushed') # execute appcontext_popped when requesting context push = _ signals.signal (' appcontext-popped') # execute message_flashed = _ signals.signal ('message-flashed') # when calling flask to add data to it Automatic trigger

Let's take a look at how signals are used in flask before the request arrives.

From flask import Flaskfrom flask.signals import _ signals, request_startedimport timeapp = Flask (_ _ name__) def wechat (args): print ('you have msg from wechat') # introduce the determined signal from flask and register a function request_started.connect (wechat) @ app.route (' /', methods= ['GET','POST'], endpoint='index') def index (): return' hello blinker'if _ name__ = ='_ main__': app.run ()

When the request arrives, flask notifies the recipient via request_started, which is the function wechat, and the wechat function executes first and then returns the result to the browser.

However, this method is not very authentic, because the signal does not support asynchronous methods, so the receiver of the signal in the production environment is usually configured with asynchronous execution framework, such as the famous asynchronous framework celery in python.

Summary

The advantages of the signal:

Decoupling applications: decompose serially running coupled applications into multi-level execution

Publish subscribers: reduce the use of callers and notify multiple subscribers at a time

Disadvantages of the signal:

Asynchrony is not supported

Limited ability to support subscription topics

At this point, the study on "how to use the signal communication blinker in python" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.