Python의 강력한 신호 라이브러리: 깜박임

1. 시그널

안녕하세요 여러분 라오비아오입니다~

시그널은 보내는 사람과 받는 사람으로 나누어집니다. 송신자는 신호를 보내고 수신자는 신호를 수신하는 프로세스를 실행한 후 다시 원래 위치로 점프하여 실행을 계속합니다.

키보드를 통해 Ctrl+C를 입력하는 Linux의 일반적인 신호는 시스템에 현재 프로세스를 종료하라고 알리는 신호를 보내는 것입니다.

신호의 특징은 발신자가 구독자에게 무슨 일이 일어났는지 알리는 것입니다. 신호 사용은 신호 정의, 신호 모니터링, 신호 전송의 3단계로 구분됩니다.

파이썬에서 신호 개념을 제공하는 통신 모듈이 블링커입니다.

Blinker는 Python 기반의 강력한 신호 라이브러리로, 간단한 지점 간 통신과 지점 간 멀티캐스트를 모두 지원합니다. Flask의 신호 메커니즘은 이를 기반으로 합니다. Blinker의 커널은 작지만 매우 강력합니다.

• 전역 명명된 신호 등록 지원
• 익명 신호 지원
• 사용자 지정 명명된 신호 지원
• 수신기와의 지속적이고 일시적인 연결 지원
• 약한 참조를 통해 자동으로 수신기와의 연결을 끊습니다.
• 모든 크기의 데이터 전송을 지원합니다.
• 신호 수신기의 반환 값 수집을 지원합니다.
• 스레드 안전성

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 익명 신호

blinker는 익명 신호도 지원합니다. 즉, 특정 신호 값을 지정할 필요가 없습니다. 생성된 각 익명 신호는 서로 독립적입니다.

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

2.3 멀티캐스트 신호

멀티캐스트 신호는 신호의 장점을 더 잘 반영할 수 있는 특성입니다. 여러 수신자가 신호에 등록되어 있으며 발신자는 여러 수신자에게 정보를 전달하기 위해 한 번만 보내면 됩니다.

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

기능 등록 외에도 더 간단한 신호 등록 방법, 즉 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 토픽을 구독할 수 있는 데코레이터

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 Simple 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 기능이 먼저 실행된 후 그 결과가 브라우저에 반환됩니다.

그러나 이 사용 방법은 신호가 비동기 방법을 지원하지 않기 때문에 그다지 정통하지 않습니다. 따라서 일반적으로 프로덕션 환경의 신호 수신자는 Python의 유명한 비동기 프레임워크 셀러리와 같이 비동기 실행으로 구성된 프레임워크입니다.

4.

신호의 장점 요약:

1. 분리된 애플리케이션: 직렬로 실행되는 결합된 애플리케이션을 다단계 실행으로 분해
2. 구독자 게시: 호출자 사용을 줄이고 한 번의 호출로 여러 구독자에게 알립니다

신호의 단점:

1. 비동기식을 지원하지 않음
2. 주제 구독 지원 기능이 제한됨

모든 좋아요, 보기, 모든 댓글, 모든 리트윗은 저에게 매우 중요합니다. 지원해 주셔서 감사합니다.

그럼 다음에 만나요. 저는 고양이와 기술 그리고 시시의 옛 사촌을 더욱 좋아합니다⁽⁽ଘ( ˙꒳˙ )ଓ⁾⁾

위 내용은 Python의 강력한 신호 라이브러리: 깜박임의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!