Flask Signal非常简单易用,能大大降低代码的耦合度,也能够让基于Flask开发的系统更加健壮。然而,虽然Flask Signal上手很容易,但它却不是Flask开发中一个入门级的功能,也不是每个开发者都会想到去使用的功能,这主要是由于不理解这个机制造成的。因此,使用Flask Signal前,首先要理解Flask Signal是什么。
什么是Signal?Flask官方文档的描述的过于言简意赅,资深开发者能够马上明白,而初学者恐怕摸不到头脑,因此Gevin对官方文档的描述做简单扩展:Signal用于解耦系统的行为和业务逻辑,这种解耦是通过 当某些行为被触发时,自动发送定义好的一种信号,与这个信号绑定的一些业务逻辑或行为,接收到这个信号后,会自动执行各自相应的业务逻辑 。与该信号绑定的业务逻辑,可以是事先预定义好的,也可以是在后续开发中随需求变动新增上去的,基于signal的机制,系统会更加稳定和易于扩展,也使得系统的业务逻辑更加清晰 —— 当某个行为触发时,发送一个信号即可,所有与该行为有关的业务逻辑或行为都会自动触发,从而实现了解耦。换句话说,Flask Signal是 观察者模式 的一种实现,它使得我们开发的系统,更加符合 开闭原则 。
『设计模式』既是编码典型问题的经典解决方案,也算是一种『行话』,不是代码初学者需要接触的内容,所以才导致了Flask Signal对资深开发者和初学者而言,在理解上出现这么大的差异。理解了Signal的本质,使用就不是问题了,如果Gevin的解释还无法让你理解,恐怕你要从『观察者模式』入手了,但Gevin建议,没有足够的代码积累,不要过早接触太多设计模式,否则无异于揠苗助长。
Flask signal的使用非常简单,Flask 官方文档 把signal介绍的过于复杂,不利于入门,在理解了signal后,只要学会以下几点便可以轻松掌握signal的使用。
(1)Signal的创建
Signal的创建只需下面3行代码即可完成:
from blinker import Namespace my_signals = Namespace() model_saved = my_signals.signal('model-saved')
(2)Signal的发送
Signal的发送通过成员函数 send()
即可完成, send()
函数的第一个参数为signal的sender,在 class
中发送signal和在 function
中传递的sender参数略有不同,class中 sender=self
, function
中 sender=current_app._get_current_object()
,即:
# In case send signal in a class: class Model(object): ... def save(self): model_saved.send(self) # In case send signal in a function: def save_model(): ... model_saved.send(current_app._get_current_object())
sender()
函数除了 sender
参数外,还可以添加多个可选参数,这些可选参数是为signal的订阅者使用的,具体例子见下一小节。
(3)编写Signal的订阅者
Signal的订阅者是一些订阅了指定signal(如上文中的 model_saved
)的函数,当signal被激活(即调用 send()
函数发送signal)时,会自动触发这些订阅者的调用,以完成某些功能。将一个普通函数变为signal的订阅函数非常简单,只要加一个 decorator
即可,仍以上面signal为例,定义一个订阅者方法如下:
@model_saved.connect_via(app) def on_model_saved(): # do something ...
对于大型项目,代码组织比较复杂,也许 app
是在系统运行时创建的,上面代码中的 app
可以用 current_app._get_current_object()
取代。
除了 connect_via
, connect
也可以,即:
@model_saved.connect def on_model_saved(): # do something ...
如果订阅者函数有参数,需要在发送signal时,将相关参数作为 signal.send()
函数的可选参数传入,这样订阅者函数可以接收到相应参数,举个详细的例子:
from blinker import Namespace from . import models # Define a signal octblog_signals = Namespace() post_visited = octblog_signals.signal('post-visited') # Define a subscriber @post_visited.connect def on_post_visited(sender, post, **extra): tracker = models.Tracker() tracker.post = post ... tracker.save() # Emit signal in a function def post_detail(slug, is_preview=False): post = models.Post.objects.get_or_404(slug=slug, post_type=post_type) data['post'] = post # do something ... # send signal if not is_preview: signals.post_visited.send(current_app._get_current_object(), post=post) ...
Flask Signal上手简单,功能强大,最关键的是它能够解耦业务和行为,使代码的逻辑更简洁,推荐使用。
稍后有时间我会在GitHub上放一个简单但完整的Flask Signal使用的小案例,如果大家有好的应用场景,欢迎给我留言~