Python の強力な信号ライブラリ: ブリンカー

1. シグナル

みなさん、こんにちは、私は私の古いいとこです~

シグナルは通知または通信の手段です。送信側と受信側。送信側がシグナルを送信し、シグナルを受信する受信側の処理が信号処理関数にジャンプし、実行後は元の位置にジャンプして実行を継続します。

Linux の一般的なシグナル、キーボードから Ctrl C を入力すると、システムにシグナルが送信され、現在のプロセスを終了するようシステムに指示されます。

シグナルの特徴は、送信者が何が起こったのかを加入者に通知することです。シグナルの使用は、シグナルの定義、シグナルの監視、シグナルの送信の 3 つのステップに分かれています。

#Python で信号の概念を提供する通信モジュールは、blinker です。

Blinker は、Python ベースの強力な信号ライブラリであり、単純なポイントツーポイント通信とポイントツーマルチポイント マルチキャストの両方をサポートします。 Flask のシグナリング メカニズムはこれに基づいています。 Blinker のカーネルは小さいながらも非常に強力で、次の機能をサポートしています:

• グローバル名前付き信号の登録をサポート
• 匿名信号をサポート
• をサポートカスタムネーミング Signal
• 受信側との永続的な接続と短期間の接続をサポートします
• 弱い参照を介した受信側との自動切断を実装します
• 任意のサイズのデータ​​の送信をサポートします
• 信号受信機の戻り値の収集をサポート
• スレッド セーフ

2. ブリンカーは

インストール方法を使用します:

pip install blinker

2.1名前付き信号

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

2.2 匿名信号

ブリンカーは匿名信号もサポートしています。つまり、特定の信号値を指定する必要はありません。作成された各匿名信号は互いに独立しています。

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

2.3 マルチキャスト信号

マルチキャスト信号は、信号の利点をよりよく反映できる特性です。複数の受信者がシグナルに登録されており、送信者は 1 回送信するだけで複数の受信者に情報を配信できます。

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 受信者はトピックをサブスクライブします

受信者は、指定されたトピックへのサブスクライブをサポートします。指定されたトピックがメッセージを送信する場合にのみ、それがサポートされます。受信者に送信されます。この方法では、さまざまなトピックを適切に区別できます。

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 デコレータの使用法

関数の登録に加えて、より簡単な信号の登録方法、つまりデコレータがあります。

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 トピックをサブスクライブできるデコレータ

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)

2.7 シグナルの受信者が存在するかどうかを確認します。

送信者がメッセージを送信する前に準備に時間がかかる場合は、受信者がいないことによるパフォーマンスの無駄を避けるため、受信機の場合、正確にするために、最初に特定の信号の受信機があるかどうかを確認し、受信機があることが確認された場合にのみ送信できます。

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 サブスクライバーが特定のシグナルにサブスクライブしているかどうかを確認する

サブスクライバーが特定のシグナルにサブスクライブしているかどうかを確認することもできます

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 シグナルは、以下に基づいています。ブリンカー

Flask は、分離されたアプリケーションのソリューションとしてブリンカーを統合します。 Flask におけるシグナルの使用シナリオは、リクエスト到着前とリクエスト終了後です。同時に、Flask はカスタムシグナルもサポートします。

3.1 簡単な 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 ウィンカーを返します。

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関数が先に実行され、結果をブラウザに返します。

しかし、この使用方法はあまり本格的ではありません。シグナルは非同期メソッドをサポートしていないためです。そのため、通常、運用環境でのシグナルの受信者は、Python の有名な非同期フレームワーク celery など、非同期実行用に構成されたフレームワークです。

4. 概要

シグナルの利点:

1. 分離されたアプリケーション: シリアル実行中の結合されたアプリケーションをマルチレベルの実行に分解します
2. パブリッシュ サブスクライバー:発信者の使用を減らし、1 回の通話で複数のサブスクライバーに通知します。

シグナルの欠点:

1. 非同期をサポートしません
2. サブスクリプションをサポートします テーマの機能は次のとおりです。限定

あなたの「いいね！」、閲覧、メッセージ、リツイートはすべて私にとって非常に重要です。ご支援に感謝します。

わかりました、また次回。猫とテクノロジーとシシィのいとこがもっと大好きです⁽⁽ଘ( ˙꒳˙ )ଓ⁾⁾

以上がPython の強力な信号ライブラリ: ブリンカーの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。