胖蔡叨叨叨
你听我说

Flask中使用Blinker实现信号传递

Blinker是为了Python提供简单快速的信号机制,其核心就是框架核心功能或者是Flask扩展发生动作时所发送的通知,用于解耦大型应用功能。Blinker目前还不是Flask的默认依赖,所以想要Flask中能使用信号机制就需要我们提前先安装Blinker依赖库:

$ pip install blinker

Blinker 介绍

Blinker Jason Kirtand 编写,并根据 MIT 许可证提供。该库支持 Python 2.7Python 3.5 或更高版本;或 Jython 2.7 或更高版本;或 PyPy 2.7 或更高版本。首先我们通过一个示例来了解下如何使用Blinker:

from blinker import signal

started =signal('test-started') # 创建信号

def each(round):
    print('Round {}!'.format(round))

def round_two(round):
    print('Only {}'.format(round))

started.connect(each)  # 订阅信号each
started.connect(round_two,sender=2) # 订阅信号round_two

for round in rang(1,4):
    started.send(round)  # 发送信号

# 运行结果
Round 1!
Round 2!
Only 2
Round 3!

如上,我们可以通过signal创建信号,通过connect来订阅信号,当订阅信息发生修改时,会触发订阅对象。通过信号的send方法触发信号。信号

匿名信号

信号不需要命名。每次调用 Signal 构造函数时都会创建一个唯一的信号。例如,上面处理器的替代实现可能会提供处理信号作为类属性:

from blinker import signal
class AltProcessor:
   on_ready = signal()
   on_complete = signal()

   def __init__(self, name):
       self.name = name

   def go(self):
       self.on_ready.send(self)
       print("Alternate processing.")
       self.on_complete.send(self)

   def __repr__(self):
       return '<AltProcessor %s>' % self.name

通过装饰器订阅信号

自从Blinker 1.1 开始可以用新的connect_via()装饰器订阅信号。使用代码如下:

from blinker import signal

started = signal('start-doc')

@started.connect_via("1")
@started.connect_via("2")
def completed(sender):
    print("AltProcessor {} completed!".format(sender))

started.send("2")

自定义信号

在Flask应用中我们可以直接使用Blinker创建信号,现在定义一种对于上传大文件的信号:

from blinker import Namespace

web_signals =  Namespace()
large_file_saved = web_signals.signal('large-file-saved')

当上传文件大于一个阈值时,可以发送这个信号。

赞(0) 打赏
转载请附上原文出处链接:胖蔡叨叨叨 » Flask中使用Blinker实现信号传递
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

请小编喝杯咖啡~

支付宝扫一扫打赏

微信扫一扫打赏