django信号系统 - 事件驱动编程模式

📂 所属阶段：第二部分 — 进阶特性
🎯 难度等级：中级
⏰ 预计学习时间：2小时
🎒 前置知识：模型设计与ORM操作

核心概念：什么是django信号？

django信号是观察者模式的标准实现，用于松耦合的应用组件通信：当某个「事件」发生时（如模型保存/删除、请求开始/结束），触发信号，自动通知所有「订阅」该信号的接收者。

信号的三个核心角色

角色说明
信号事件的抽象载体（如 post_save）
发送者触发信号的主体（如某个具体的 User 模型类）
接收者响应信号的函数/类方法（如「发送欢迎邮件」）

快速上手：最基础的信号使用

1. 引入依赖

from django.dispatch import Signal, receiver
from django.db.models.signals import post_save
from myapp.models import User

2. 接收者注册（两种方式）

方式一：装饰器语法（推荐）

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    """新用户注册成功后发送欢迎邮件"""
    if created:
        # 实际项目请用Celery异步
        print(f"向 {instance.email} 发送欢迎邮件")

方式二：手动连接

def create_user_profile(sender, instance, created, **kwargs):
    """新用户注册成功后创建档案"""
    if created:
        from myapp.models import UserProfile
        UserProfile.objects.create(user=instance)

# 在 apps.py 的 ready() 中调用此代码
post_save.connect(create_user_profile, sender=User)

3. 信号触发（无需写额外代码）

当执行 User.objects.create_user(...) 或 user.save() 时，post_save 信号会自动触发。

常用内置信号

django内置了三类常用信号，覆盖核心场景。

1. 模型信号（最常用）

操作数据库模型时自动触发：

from django.db.models.signals import (
    pre_init,    # 模型实例化前（几乎不用）
    post_init,   # 模型实例化后
    pre_save,    # 模型保存前
    post_save,   # 模型保存后（★核心：区分created/更新）
    pre_delete,  # 模型删除前
    post_delete, # 模型删除后
    m2m_changed, # 多对多关系变更
)

模型信号示例：自动更新文章字数

from django.db.models.signals import pre_save
from django.dispatch import receiver
from myapp.models import Article

@receiver(pre_save, sender=Article)
def auto_count_words(sender, instance, **kwargs):
    """文章保存前自动统计字数"""
    if instance.content:
        instance.word_count = len(instance.content.strip().split())

2. 请求信号

处理HTTP请求时触发：

from django.core.signals import (
    request_started,    # 请求开始
    request_finished,   # 请求结束
    got_request_exception, # 请求异常（可用于全局错误记录）
)

3. 应用/项目信号

from django.db.models.signals import (
    pre_migrate,  # 迁移执行前
    post_migrate, # 迁移执行后（★核心：创建初始数据）
)

应用信号示例：迁移后创建超级管理员

from django.db.models.signals import post_migrate
from django.dispatch import receiver
from django.contrib.auth.models import User

@receiver(post_migrate)
def create_superuser(app_config, **kwargs):
    """仅在myapp迁移完成后创建超级管理员"""
    if app_config.name == 'myapp':
        if not User.objects.filter(username='admin').exists():
            User.objects.create_superuser(
                'admin', 'admin@example.com', 'admin123'
            )

自定义信号：满足个性化需求

当内置信号不够用时，可以创建自定义信号。

1. 定义自定义信号

# myapp/signals.py
from django.dispatch import Signal

# 定义一个「订单完成」的信号，携带order、user参数
order_completed = Signal()

2. 发送自定义信号

# myapp/services.py
from myapp.models import Order
from myapp.signals import order_completed

def complete_order(order_id):
    """标记订单完成并发送信号"""
    order = Order.objects.get(id=order_id)
    order.status = 'COMPLETED'
    order.save()
    
    # 发送信号（★显式调用）
    order_completed.send(
        sender=Order,
        order=order,
        user=order.user
    )

3. 接收自定义信号

# myapp/signals.py
from django.dispatch import receiver
from .models import Order

@receiver(order_completed, sender=Order)
def send_order_confirmation(sender, order, user, **kwargs):
    """订单完成后发送确认通知"""
    print(f"向 {user.email} 发送订单 {order.id} 的确认")

最佳实践与避坑指南

⚠️ 避坑1：信号必须在应用启动时被导入

如果信号模块没被导入，接收者不会生效！必须在 apps.py 的 ready() 中显式导入：

# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        # 显式导入信号模块，触发装饰器的注册逻辑
        import myapp.signals

⚠️ 避坑2：默认是同步阻塞的

信号接收者如果执行耗时操作（如发邮件、调用API），会直接阻塞主线程！必须用异步任务（如Celery）：

# myapp/signals.py
from celery import shared_task

@shared_task
def async_send_welcome_email(user_id):
    """异步发送欢迎邮件"""
    from django.contrib.auth.models import User
    user = User.objects.get(id=user_id)
    print(f"异步向 {user.email} 发送邮件")

@receiver(post_save, sender=User)
def trigger_async_email(sender, instance, created, **kwargs):
    """只触发异步任务，不阻塞"""
    if created:
        async_send_welcome_email.delay(instance.id)

⚠️ 避坑3：避免信号循环调用

如果在 post_save 接收者中再次调用 instance.save()，会触发无限循环！解决方法有两个：

  1. 添加处理标志
@receiver(post_save, sender=Article)
def safe_article_update(sender, instance, created, **kwargs):
    if getattr(instance, '_in_signal', False):
        return
    instance._in_signal = True
    try:
        # 安全的更新逻辑
        if not instance.slug:
            instance.slug = f"article-{instance.id}"
            instance.save()
    finally:
        instance._in_signal = False
  1. 使用 update_fields 或 QuerySet.update()
@receiver(post_save, sender=Article)
def safe_article_update(sender, instance, created, **kwargs):
    if not instance.slug and created:
        # QuerySet.update() 不会触发 post_save
        Article.objects.filter(id=instance.id).update(slug=f"article-{instance.id}")

⚠️ 避坑4：大量数据操作时临时断开信号

bulk_create 不会触发单个信号，但如果必须逐行操作，记得临时断开信号以提升性能：

from contextlib import contextmanager
from django.db.models.signals import post_save

@contextmanager
def temp_disconnect_signal(signal, receiver, sender):
    """临时断开信号的上下文管理器"""
    signal.disconnect(receiver, sender=sender)
    try:
        yield
    finally:
        signal.connect(receiver, sender=sender)

# 使用示例
with temp_disconnect_signal(post_save, send_welcome_email, User):
    # 批量导入1000个测试用户，不会触发欢迎邮件
    for i in range(1000):
        User.objects.create_user(f"test{i}", f"test{i}@example.com", "123")

本章小结

django信号是解耦应用组件的利器，但必须谨慎使用：

  1. ✅ 适合场景：数据同步、缓存失效、审计日志、通知推送
  2. ❌ 不适合场景：核心业务逻辑（信号是「可选的通知」，不是「必须的依赖」）
  3. 🚀 优化建议：用异步任务、避免循环、大量操作时断开信号

🔗 相关教程推荐

🏷️ 标签云: django信号 事件驱动 观察者模式 解耦 Celery 性能优化