Django 信号

概述

Django信号系统是一个基于观察者模式的事件系统，允许在特定事件发生时自动执行相关代码。它实现了应用组件之间的解耦，使得代码更加模块化和可维护。

核心概念

1. 信号（Signal）

信号是Django中的事件通知机制，当某个动作发生时，信号会被发送给所有注册的接收器。

2. 发送器（Sender）

发送信号的对象，通常是Django模型类或None（表示任何发送器）。

3. 接收器（Receiver）

响应信号的函数或方法，当信号被发送时会被自动调用。

4. 连接（Connect）

将接收器注册到特定信号的过程。

工作原理

基本流程

1
2
3
4
5
1. 注册阶段（connect）
   用户定义处理函数 → 连接到信号 → 存储到接收器列表

2. 触发阶段（send）
   事件发生 → Django发送信号 → 查找接收器 → 调用处理函数

核心机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 简化的信号系统实现
class SimpleSignal:
    def __init__(self):
        self.receivers = []  # 存储 (lookup_key, receiver_ref) 对
        
    def connect(self, receiver, sender=None):
        # 1. 生成查找键
        lookup_key = (id(receiver), id(sender))
        # 2. 创建弱引用
        receiver_ref = weakref.ref(receiver)
        # 3. 存储
        self.receivers.append((lookup_key, receiver_ref))
        
    def send(self, sender, **kwargs):
        # 1. 查找匹配的接收器
        # 2. 从弱引用获取真实函数
        # 3. 调用函数并返回结果
        results = []
        for key, ref in self.receivers:
            if key[1] == id(sender):
                receiver = ref()  # 获取真实函数
                if receiver:
                    result = receiver(sender=sender, **kwargs)
                    results.append((receiver, result))
        return results

Django内置信号

模型信号

pre_save / post_save

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from django.db.models.signals import pre_save, post_save
from django.contrib.auth.models import User

def user_pre_save(sender, instance, **kwargs):
    print(f"即将保存用户: {instance.username}")

def user_post_save(sender, instance, created, **kwargs):
    if created:
        print(f"新用户创建: {instance.username}")
    else:
        print(f"用户更新: {instance.username}")

pre_save.connect(user_pre_save, sender=User)
post_save.connect(user_post_save, sender=User)

pre_delete / post_delete

1
2
3
4
5
6
7
8
9
10
from django.db.models.signals import pre_delete, post_delete

def user_pre_delete(sender, instance, **kwargs):
    print(f"即将删除用户: {instance.username}")

def user_post_delete(sender, instance, **kwargs):
    print(f"用户已删除: {instance.username}")

pre_delete.connect(user_pre_delete, sender=User)
post_delete.connect(user_post_delete, sender=User)

m2m_changed

1
2
3
4
5
6
7
8
9
from django.db.models.signals import m2m_changed

def groups_changed(sender, instance, action, pk_set, **kwargs):
    if action == "post_add":
        print(f"用户 {instance.username} 加入了新组")
    elif action == "post_remove":
        print(f"用户 {instance.username} 离开了组")

m2m_changed.connect(groups_changed, sender=User.groups.through)

请求/响应信号

request_started / request_finished

1
2
3
4
5
6
7
8
9
10
from django.core.signals import request_started, request_finished

def request_start_handler(sender, environ, **kwargs):
    print("请求开始处理")

def request_finish_handler(sender, **kwargs):
    print("请求处理完成")

request_started.connect(request_start_handler)
request_finished.connect(request_finish_handler)

实现细节

1. 数据结构

1
2
3
4
5
class Signal:
    def __init__(self):
        self.receivers = []  # [(lookup_key, receiver_ref), ...]
        self.lock = threading.RLock()  # 线程安全锁
        self.sender_receivers_cache = weakref.WeakKeyDictionary()  # 性能缓存

2. 连接器实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def connect(self, receiver, sender=None, weak=True, dispatch_uid=None):
    # 参数验证
    if settings.DEBUG:
        assert callable(receiver), "接收器必须是可调用对象"
        if not func_accepts_kwargs(receiver):
            raise ValueError("接收器必须接受关键字参数")
    
    # 生成查找键
    if dispatch_uid:
        lookup_key = (dispatch_uid, _make_id(sender))
    else:
        lookup_key = (_make_id(receiver), _make_id(sender))
    
    # 处理弱引用
    if weak:
        if hasattr(receiver, '__self__') and hasattr(receiver, '__func__'):
            receiver_ref = WeakMethod(receiver)  # 绑定方法特殊处理
        else:
            receiver_ref = weakref.ref(receiver, self._remove_receiver)
    else:
        receiver_ref = receiver
    
    # 线程安全地添加
    with self.lock:
        # 检查重复
        for r_key, _ in self.receivers:
            if r_key == lookup_key:
                return  # 已存在
        
        # 添加新接收器
        self.receivers.append((lookup_key, receiver_ref))
        self.sender_receivers_cache.clear()  # 清空缓存

3. 发送器实现

1
2
3
4
5
6
7
8
9
10
def send(self, sender, **named):
    # 快速检查
    if not self.receivers or self.sender_receivers_cache.get(sender) is NO_RECEIVERS:
        return []
    
    # 获取活跃接收器并调用
    return [
        (receiver, receiver(signal=self, sender=sender, **named))
        for receiver in self._live_receivers(sender)
    ]

4. 弱引用机制

为什么使用弱引用？

1
2
3
4
5
6
7
8
9
10
11
12
# 问题：强引用导致内存泄漏
class MyClass:
    def handler(self):
        pass

obj = MyClass()
signal.connect(obj.handler)  # 强引用阻止obj被回收
del obj  # obj无法被删除，造成内存泄漏

# 解决：弱引用允许正常回收
signal.connect(obj.handler, weak=True)  # 使用弱引用
del obj  # obj可以正常被回收

绑定方法的特殊处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 普通函数
def my_function():
    pass
ref = weakref.ref(my_function)  # 正常工作

# 绑定方法的问题
class MyClass:
    def my_method(self):
        pass

obj = MyClass()
method = obj.my_method  # 这是一个临时对象
ref = weakref.ref(method)  # 会立即失效！

# 解决方案：使用WeakMethod
ref = WeakMethod(obj.my_method)  # 正确处理绑定方法

5. 性能优化

缓存机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def _live_receivers(self, sender):
    # 检查缓存
    receivers = self.sender_receivers_cache.get(sender)
    if receivers is None:
        # 缓存未命中，重新构建
        receivers = []
        sender_id = _make_id(sender)
        
        for (r_key, r_ref) in self.receivers:
            if r_key[1] == sender_id or r_key[1] == _make_id(None):
                receiver = self._get_receiver(r_ref)
                if receiver:
                    receivers.append(receiver)
        
        # 缓存结果
        if receivers:
            self.sender_receivers_cache[sender] = receivers
        else:
            self.sender_receivers_cache[sender] = NO_RECEIVERS
    
    return receivers

早期退出

1
2
3
4
5
6
7
8
9
def send(self, sender, **named):
    # 两层检查，快速退出
    if not self.receivers:
        return []  # 没有任何接收器
    
    if self.sender_receivers_cache.get(sender) is NO_RECEIVERS:
        return []  # 该发送器没有接收器
    
    # 继续处理...

使用模式

1. 基本用法

1
2
3
4
5
6
7
8
9
10
11
from django.db.models.signals import post_save
from django.contrib.auth.models import User

def welcome_new_user(sender, instance, created, **kwargs):
    if created:
        print(f"欢迎新用户: {instance.username}")
        # 发送欢迎邮件
        # 创建用户配置文件
        # 记录日志

post_save.connect(welcome_new_user, sender=User)

2. 使用装饰器

1
2
3
4
5
6
from django.dispatch import receiver

@receiver(post_save, sender=User)
def user_post_save_handler(sender, instance, created, **kwargs):
    if created:
        print(f"新用户创建: {instance.username}")

3. 防止重复连接

1
2
3
4
5
6
7
8
9
10
11
12
post_save.connect(
    welcome_new_user, 
    sender=User, 
    dispatch_uid="welcome_new_user_unique"
)

# 即使多次调用，也只会连接一次
post_save.connect(
    welcome_new_user, 
    sender=User, 
    dispatch_uid="welcome_new_user_unique"  # 相同ID，不会重复
)

4. 手动断开连接

1
2
3
4
5
6
7
8
# 连接
post_save.connect(my_handler, sender=User)

# 断开
post_save.disconnect(my_handler, sender=User)

# 使用dispatch_uid断开
post_save.disconnect(sender=User, dispatch_uid="my_unique_handler")

5. 自定义信号

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import django.dispatch

# 定义自定义信号
user_logged_in = django.dispatch.Signal()

# 连接处理器
@receiver(user_logged_in)
def handle_user_login(sender, user, ip_address, **kwargs):
    print(f"用户 {user.username} 从 {ip_address} 登录")

# 在视图中发送信号
def login_view(request):
    # ... 登录逻辑 ...
    user_logged_in.send(
        sender=None,
        user=request.user,
        ip_address=request.META.get('REMOTE_ADDR')
    )

错误处理

send vs send_robust

send方法（快速失败）

1
2
3
4
5
6
def send(self, sender, **named):
    # 任何接收器抛出异常都会中断整个过程
    return [
        (receiver, receiver(signal=self, sender=sender, **named))
        for receiver in self._live_receivers(sender)
    ]

send_robust方法（容错处理）

1
2
3
4
5
6
7
8
9
10
def send_robust(self, sender, **named):
    # 捕获异常，继续执行其他接收器
    responses = []
    for receiver in self._live_receivers(sender):
        try:
            response = receiver(signal=self, sender=sender, **named)
        except Exception as err:
            response = err
        responses.append((receiver, response))
    return responses

最佳实践

1. 接收器设计原则

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def good_receiver(sender, instance, created, **kwargs):
    """
    良好的接收器设计：
    1. 接受**kwargs参数
    2. 处理业务逻辑简洁
    3. 避免修改传入的对象
    4. 异常处理得当
    """
    if created:
        try:
            # 业务逻辑
            send_welcome_email(instance)
        except Exception as e:
            logger.error(f"发送欢迎邮件失败: {e}")
            # 不要重新抛出异常，除非必要

2. 避免循环调用

1
2
3
4
5
6
7
8
9
10
11
12
# 危险：可能导致无限递归
@receiver(post_save, sender=User)
def update_user_profile(sender, instance, **kwargs):
    instance.last_updated = timezone.now()
    instance.save()  # 这会再次触发post_save信号！

# 安全：使用update避免触发信号
@receiver(post_save, sender=User)
def update_user_profile(sender, instance, **kwargs):
    User.objects.filter(pk=instance.pk).update(
        last_updated=timezone.now()
    )  # update不会触发信号

3. 性能考虑

1
2
3
4
5
6
7
8
9
10
# 避免在信号处理器中执行耗时操作
@receiver(post_save, sender=User)
def heavy_task_handler(sender, instance, created, **kwargs):
    if created:
        # 不好：同步执行耗时任务
        send_complex_email(instance)  # 可能需要几秒钟
        
        # 好：异步执行
        from celery import current_app
        current_app.send_task('send_welcome_email', [instance.id])

4. 测试信号

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from django.test import TestCase
from django.db.models.signals import post_save

class SignalTestCase(TestCase):
    def test_user_creation_signal(self):
        # 方法1：检查信号是否被正确发送
        with self.assertSignalSent(post_save, sender=User):
            User.objects.create_user(username='test')
        
        # 方法2：临时断开信号进行测试
        post_save.disconnect(welcome_new_user, sender=User)
        try:
            user = User.objects.create_user(username='test')
            # 测试不依赖信号的逻辑
        finally:
            post_save.connect(welcome_new_user, sender=User)

总结

Django信号系统的核心机制：

  1. 注册阶段：connect方法将接收器函数与信号关联，存储为弱引用
  2. 触发阶段：send方法根据发送者查找对应接收器并调用
  3. 内存管理：弱引用确保对象可以正常被垃圾回收
  4. 性能优化：缓存机制和早期退出提高执行效率
  5. 线程安全：使用锁保护共享数据结构

信号系统实现了松耦合的事件驱动架构，是Django框架中优雅的设计模式之一。正确使用信号可以让代码更加模块化和可维护。

#Django
Django 信号
https://luffy997.github.io/2025/07/01/Django-信号/
作者
Luffy997
发布于
2025年7月1日
许可协议