最近，我负责开发一个重后端的应用。这个应用数据流向复杂，处理逻辑冗余堆积。项目技术栈选择的是 Django + Vuejs。前端使用 Webpack 打包，模块化管理，主要是展示数据。后端涉及的模块多，处理规则多，数据表多，每次涉及之前功能修改时，都消耗大量时间 review 代码。这让我意识到，在复杂应用中，解耦模块非常重要。下面是一些思考和实践。
1. 观察者模式
在实践中，我主要使用的是 Django Signal，实现对模块的解耦。Django Signal 是 Django 对观察者模式的实现和应用。因此，有必要先了解一下观察者模式。观察者模式是软件设计模式的一种。通常，大家会使用等式：发布 + 订阅 = 观察者模式。来表达对观察者模式的理解。实际上，这个等式并不完全正确。发布订阅模式与观察者模式区别：
- 发布订阅模式的通信依赖于消息队列（RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等），属于异步；观察者模式通常是同步的
- 发布订阅模式松散耦合，发布者和订阅者甚至所属不同应用；观察者模式所属一个应用
在实现上，观察者模式，需要维护一个订阅列表。当状态发生改变时，自动通知列表中的全部对象。
2. Django Signal
Signal 是 Django 框架中提供的一个信号分发器。发送器发送信号，通知一系列的接收器，从而触发接收器执行一些操作。需要注意的是，Django 信号是同步的。如果滥用，会影响到 Django 的处理效率。下面我会以 Django 1.8.3 为例，从一个使用案例出发，再到源码，介绍 Django 中 Signal 的实现方式。
2.1 一个简单的使用案例
这里有一个小需求：在 MyModel 表执行 save 后，触发一些执行逻辑。
myApp/__init__.py
1
2
|
# -*- coding: utf-8 -*-
default_app_config = 'myApp.apps.MyAppConfig'
|
myApp/apps.py
1
2
3
4
5
6
7
8
9
|
# -*- coding: utf-8 -*-
from django.apps import AppConfig
class MyAppConfig(AppConfig):
name = 'myApp'
def ready(self):
import myApp.signals.handlers
|
myApp/signals/handlers.py
1
2
3
4
5
6
7
8
9
10
11
|
# -*- coding: utf-8 -*-
from django.dispatch import receiver
from django.db.models.signals import post_save
from myApp.models import MyModel
@receiver(post_save, sender=MyModel, dispatch_uid="mymodel_post_save")
def my_model_handler(sender, **kwargs):
# 这里写 MyModel 执行 save 后的逻辑
pass
|
2.2 从源码理解 Django Signal 处理逻辑
上面的例子，使用了极少量的代码，就享受到了 Django 提供的信号处理机制所带来的便利。但是，如果仅仅停留在使用，你可能无法对 Django Signal 有更深入的了解。下面，从源码来看看 Django Signal 的处理逻辑。
Django 内置了大量 Model 相关的信号，可以直接使用。上面例子使用的信号 post_save ，就是 ModelSignal 类的一个实例，而 ModelSignal 又继承自 Signal 类。django/db/models/signal.py
1
2
3
4
5
6
7
8
9
|
from django.dispatch import Signal
class ModelSignal(Signal):
def connect(self, receiver, sender=None, weak=True, dispatch_uid=None):
super(ModelSignal, self).connect(
receiver, sender=sender, weak=weak, dispatch_uid=dispatch_uid
)
post_save = ModelSignal(providing_args=["instance", "raw", "created", "using", "update_fields"], use_caching=True)
|
Django 提供的 receiver 函数是一个装饰器，被修饰的函数作为参数注册到接收器对象列表。django/dispatch/__init__.py
1
|
from django.dispatch.dispatcher import Signal, receiver
|
django/dispatch/dispatcher.py
1
2
3
4
5
6
7
8
9
|
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/dispatch/dispatcher.py
1
2
3
4
5
|
class Signal(object):
def __init__(self, providing_args=None, use_caching=False):
self.receivers = []
def connect(self, receiver, sender=None, weak=True, dispatch_uid=None):
self.receivers.append((lookup_key, receiver))
|
在 save 完成之后，Django 会主动发出 post_save 信号；如果是自定义信号，那么需要自行触发。。django/db/models/base.py
1
2
3
4
5
6
7
|
class Model(six.with_metaclass(ModelBase)):
# 触发 Model 相关的信号
def save_base(self, raw=False, force_insert=False,
force_update=False, using=None, update_fields=None):
# Signal that the save is complete
signals.post_save.send(sender=origin, instance=self, created=(not updated),
update_fields=update_fields, raw=raw, using=using)
|
处理信号，实际上就是依次调用接受器列表中的函数。django/dispatch/dispatcher.py
1
2
3
4
5
6
7
8
|
class Signal(object):
def send(self, sender, **named):
responses = []
for receiver in self._live_receivers(sender):
response = receiver(signal=self, sender=sender, **named)
responses.append((receiver, response))
return responses
|
3. 信号解耦，任务异步
在学习了观察者模式，了解 Django Signal 之后，就基本掌握了 Django 模块解耦的基础知识。接着，需要进一步明确模块之间的耦合机制，制定项目约定，就可以利落地实践了。梳理一下请求的处理链路：
请求经过接入层、中间件处理之后，由 URL 分发器匹配到合适的处理模块，最终某个模块负责返回响应。各个模块连接数据库、消息队列、对象存储保存状态。每个模块包含四部分：
- AppLogic，模块的应用逻辑
- Signal，模块内置的信号
- SignalHandle，模块关注的信号处理句柄
- CeleryTasks，模块的异步任务
模块与模块之前完全通过信号耦合：
由于 Django Signal 是同步处理机制，为了支持异步处理，可以结合 Celery 和 RabbitMQ 进行实践。下面是一个信号处理异步逻辑的例子：myApp/tasks.py
1
2
3
4
5
6
7
|
# -*- coding: utf-8 -*-
from celery import task
@task(ignore_result=True)
def my_task(instance):
pass
|
myApp/signals/handlers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
# -*- coding: utf-8 -*-
from django.dispatch import receiver
from django.db.models.signals import post_save
from myApp.models import MyModel
from myApp.tasks import my_task
@receiver(post_save, sender=MyModel, dispatch_uid="mymodel_post_save")
def my_model_handler(sender, **kwargs):
instance = kwargs['instance']
# 异步
my_task.apply_async(args=[instance])
# 同步
pass
|
