一文打尽 Python-web 开发的 signal 机制
signal在flask/django中都是很重要的解耦手段。flask的signal依赖blinker实现，django的signal也很类似。blinker库是纯python实现的代码简单，功能强大的signal库。本文我们从blinker开始，一起了解python-web开发的signal机制：
- blinker的api
- blinker-signal的实现
- flask-signal的实现
- django-signal的实现
- weakref介绍
- 小结
- 小技巧
blinker简介
blinker源码使用 1.4 版本, 项目结构如下:
| 文件 | 描述 |
|---|---|
| base.py | 核心逻辑 |
| _saferef.py | 安全引用相关逻辑 |
| _utilities.py | 工具类 |
blinker的API
blinker的api使用示例:
from blinker import signal
def subscriber1(sender):
print("1 Got a signal sent by %r" % sender)
def subscriber2(sender):
print("2 Got a signal sent by %r" % sender)
ready = signal('ready')
print(ready)
ready.connect(subscriber1)
ready.connect(subscriber2)
ready.send("go")
示例的日志输出:
<blinker.base.NamedSignal object at 0x7f93a805ad00; 'ready'>
1 Got a signal sent by 'go'
2 Got a signal sent by 'go'
可以看到signal是发布/订阅模式。或者换个更常见的说法，事件中心：
ready = signal('ready')创建名为ready的事件中心ready.connect(subscriber1)给ready事件中心添加事件监听器ready.send("go")向ready事件中心派发事件，这样事件监听器会收到事件并进行处理
signal的实现
signal默认单例，提供开箱即用的API:
class NamedSignal(Signal):
"""A named generic notification emitter."""
def __init__(self, name, doc=None):
Signal.__init__(self, doc)
self.name = name
class Namespace(dict):
def signal(self, name, doc=None):
try:
return self[name]
except KeyError:
return self.setdefault(name, NamedSignal(name, doc))
signal = Namespace().signal
需要说明一下的是，signal的单例是和name绑定的。同一个名称得到同一个NamedSignal对象，不同名称得到的NamedSignal对象不一样。
NamedSignal的父类Signal的构造方法，包括1）事件接收器字典receivers：以事件接收器id为key和事件接收器为value；2）接收器ID-发送器ID的字典：以接收器ID为key和发送器ID集合为value；3）和2类似的字典，只不过是反向的，key为发送器ID，value为接收器集合。
ANY = symbol('ANY')
class Signal(object):
ANY = ANY
def __init__(self, doc=None)
self.receivers = {}
self._by_receiver = defaultdict(set)
self._by_sender = defaultdict(set)
...
Signal的connect函数添加消息接收器，可以看到sender和receiver是多对多的关系。
def connect(self, receiver, sender=ANY, weak=True):
receiver_id = hashable_identity(receiver)
receiver_ref = receiver
sender_id = ANY_ID
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
return receiver
Signal的send函数将消息发送给所有关注该sender的接收器：
def send(self, *sender, **kwargs):
sender = sender[0]
# 循环执行所有的receiver
return [(receiver, receiver(sender, **kwargs))
for receiver in self.receivers_for(sender)]
def receivers_for(self, sender):
sender_id = hashable_identity(sender)
# 根据sender_id找receiver_id
if sender_id in self._by_sender:
# 2个set的合集
ids = (self._by_sender[ANY_ID] |
self._by_sender[sender_id])
else:
ids = self._by_sender[ANY_ID].copy()
for receiver_id in ids:
receiver = self.receivers.get(receiver_id)
if receiver is None:
continue
# 迭代器
yield receiver
有始有终，Signal使用disconnect函数注销消息的接收器：
def disconnect(self, receiver, sender=ANY):
sender_id = ANY_ID
receiver_id = hashable_identity(receiver)
self._disconnect(receiver_id, sender_id)
def _disconnect(self, receiver_id, sender_id):
if sender_id == ANY_ID:
if self._by_receiver.pop(receiver_id, False):
for bucket in self._by_sender.values():
bucket.discard(receiver_id)
self.receivers.pop(receiver_id, None)
else:
self._by_sender[sender_id].discard(receiver_id)
self._by_receiver[receiver_id].discard(sender_id)
为了便于理解signal机制，我们暂时忽略了weakref相关的代码，稍后再进行介绍。
flask-signal的实现
flask-signal依赖blinker的实现:
# flask.signals.py
from blinker import Namespace
_signals = Namespace()
template_rendered = _signals.signal("template-rendered")
before_render_template = _signals.signal("before-render-template")
request_started = _signals.signal("request-started")
request_finished = _signals.signal("request-finished")
request_tearing_down = _signals.signal("request-tearing-down")
got_request_exception = _signals.signal("got-request-exception")
appcontext_tearing_down = _signals.signal("appcontext-tearing-down")
appcontext_pushed = _signals.signal("appcontext-pushed")
appcontext_popped = _signals.signal("appcontext-popped")
message_flashed = _signals.signal("message-flashed")
从上面代码可以看到flask使用blinker预制了多个signal。以request_started为例, flask在处理request时候会向request_started派发事件:
# flask.app.py
from .signals import request_started
def full_dispatch_request(self):
...
request_started.send(self)
...
我们可以在自己的代码中，这样注册事件监听:
def log_request(sender, **extra):
sender.logger.debug('Request context is set up')
from flask import request_started
request_started.connect(log_request, app)
这样就可以很方便的使用signal获取到flask在各个阶段的数据。
django-signal的实现
django-signal虽然是独立实现，但是模式和blinker非常类似。Signal构造函数创建了一个对象，充当事件中心。
# django/dispatch/dispatcher.py
def _make_id(target):
if hasattr(target, '__func__'):
return (id(target.__self__), id(target.__func__))
return id(target)
NONE_ID = _make_id(None)
# A marker for caching
NO_RECEIVERS = object()
class Signal:
def __init__(self, providing_args=None, use_caching=False):
"""
Create a new signal.
"""
self.receivers = []
self.lock = threading.Lock()
self.use_caching = use_caching
self.sender_receivers_cache = weakref.WeakKeyDictionary() if use_caching else {}
self._dead_receivers = False
connect核心功能就是为事件监听器构建唯一标识(receiver_id,sender_id),然后加入receivers数组。
def connect(self, receiver, sender=None, weak=True, dispatch_uid=None):
from django.conf import settings
lookup_key = (_make_id(receiver), _make_id(sender))
ref = weakref.ref
receiver_object = receiver
receiver = ref(receiver)
with self.lock:
if not any(r_key == lookup_key for r_key, _ in self.receivers):
self.receivers.append((lookup_key, receiver))
send函数和blinker的send类似：
def send(self, sender, **named):
return [
(receiver, receiver(signal=self, sender=sender, **named))
for receiver in self._live_receivers(sender)
]
def _live_receivers(self, sender):
with self.lock:
senderkey = _make_id(sender)
receivers = []
for (receiverkey, r_senderkey), receiver in self.receivers:
if r_senderkey == NONE_ID or r_senderkey == senderkey:
receivers.append(receiver)
...
non_weak_receivers = []
for receiver in receivers:
non_weak_receivers.append(receiver)
return non_weak_receivers
django-signal额外提供了一个receiver装饰器，方便业务使用:
def receiver(signal, **kwargs):
def _decorator(func):
if isinstance(signal, (list, tuple)):
for s in signal:
s.connect(func, **kwargs)
else:
signal.connect(func, **kwargs)
return func
return _decorator
django的model中额外包装了ModelSignal类并且预制了一些signal:
class ModelSignal(Signal):
def _lazy_method(self, method, apps, receiver, sender, **kwargs):
from django.db.models.options import Options
# This partial takes a single optional argument named "sender".
partial_method = partial(method, receiver, **kwargs)
if isinstance(sender, str):
apps = apps or Options.default_apps
apps.lazy_model_operation(partial_method, make_model_tuple(sender))
else:
return partial_method(sender)
def connect(self, receiver, sender=None, weak=True, dispatch_uid=None, apps=None):
self._lazy_method(
super().connect, apps, receiver, sender,
weak=weak, dispatch_uid=dispatch_uid,
)
...
# 定义模型各个阶段的signal
pre_init = ModelSignal(use_caching=True)
post_init = ModelSignal(use_caching=True)
pre_save = ModelSignal(use_caching=True)
post_save = ModelSignal(use_caching=True)
pre_delete = ModelSignal(use_caching=True)
post_delete = ModelSignal(use_caching=True)
m2m_changed = ModelSignal(use_caching=True)
pre_migrate = Signal()
signal的使用方式在receiver装饰器的注释中有介绍:
"""
A decorator for connecting receivers to signals. Used by passing in the
signal (or list of signals) and keyword arguments to connect::
@receiver(post_save, sender=MyModel)
def signal_receiver(sender, **kwargs):
...
@receiver([post_save, post_delete], sender=MyModel)
def signals_receiver(sender, **kwargs):
...
"""
这样利用signal机制，可以对MyModel进行一些额外的逻辑处理，又避免了代码的硬耦合。
weakref 介绍
了解了signal的各种实现和使用后，我们再回头学习blinker-signal中另外一个环节weakref。weakref可以显著提高signal的性能, 请看下面示例:
def test_weak_value_dict(cache):
c_list = []
class C:
def method(self):
return ("method called!", id(self))
c1 = C()
c2 = C()
c3 = C()
c_list.append(c1)
c_list.append(c2)
c_list.append(c3)
del c1, c2, c3
def do_cache(cache, name, target):
cache[name] = target
for idx, target in enumerate(c_list):
do_cache(cache, idx, target)
for k, v in cache.items():
print("before", k, v.method())
del c_list
gc.collect()
for x, y in cache.items():
print("after", x, y.method())
test_weak_value_dict({})
print("==" * 10)
test_weak_value_dict(weakref.WeakValueDictionary())
在test_weak_value_dict函数中，创建了3个对象，将对象放到一个列表和cache中，完成后再删除对象和对象列表并进行gc。如果cache的实现是set，那么gc后cache中任然存在3个对象，也就是对象不会回收；如果是使用WeakValueDictionary实现的cache，则部分对象进行了回收。在一个事件中心，如果监听函数取消后却无法释放回收，内存会持续增长。
before 0 ('method called!', 140431874960640)
before 1 ('method called!', 140431874959440)
before 2 ('method called!', 140431874959968)
after 0 ('method called!', 140431874960640)
after 1 ('method called!', 140431874959440)
after 2 ('method called!', 140431874959968)
====================
before 0 ('method called!', 140431875860416)
before 1 ('method called!', 140431875860128)
before 2 ('method called!', 140431876163136)
after 2 ('method called!', 140431876163136)signal 小结
到这里我们可以知道blinker/flask/django的signal都是单纯的python消息中心，和我们之前在gunicorn中使用的系统 signal 完全不一样。消息中心，可以用来进行业务逻辑的解耦，一般就包括三步:
- 注册监听器
- 派发事件
- 注销监听器
小技巧
blinker中提供了一种 单例模式 的实现参考，我把它叫做 分组单例 , 组名相同会得到同一个对象实例:
class _symbol(object):
def __init__(self, group):
"""Construct a new group symbol."""
# 原文是name，我把它换成了group，感觉这样更容易理解一些
self.__group__ = self.group = group
def __reduce__(self):
return symbol, (self.group,)
def __repr__(self):
return self.group
_symbol.__group__ = 'symbol'
class symbol(object):
"""A constant symbol.
# group相同的symbol是同一个对象
>>> symbol('foo') is symbol('foo')
True
>>> symbol('foo')
foo
"""
symbols = {}
def __new__(cls, group):
try:
return cls.symbols[group]
except KeyError:
return cls.symbols.setdefault(group, _symbol(group))
ANY = symbol('ANY') # 单例
参考链接:
- https://pythonhosted.org/blinker/
- https://stackify.com/python-garbage-collection/
- https://www.cnblogs.com/TM0831/p/10599716.html
- https://www.geeksforgeeks.org/weak-references-in-python/
- https://pymotw.com/2/weakref/
Python实战 | 基于 Flask 部署 Keras 深度学习模型
评论