Django 信号机制（Signals）实战指南

预计阅读时间：12 分钟

Django 的信号机制（Signals）是一套内置的发布-订阅系统，允许某些发送者在特定事件发生时通知一组接收者，而发送者和接收者之间完全解耦，互不知晓对方的存在。

合理使用信号可以让代码更整洁——把"某件事发生后需要做的副作用"从核心逻辑中剥离出去，而不是把所有逻辑堆在 save() 或视图函数里。

信号的基本概念

Django 信号的三个核心角色：

Signal：信号对象本身，定义了事件类型
Sender：触发信号的发送者（通常是某个 Model 或 View）
Receiver：接收信号并执行逻辑的函数

基本工作流程：

某个事件发生
    ↓
Signal.send(sender, **kwargs)
    ↓
所有已注册的 receiver 函数依次被调用

Django 内置信号

Django 自带了一批常用信号，覆盖 Model 生命周期、请求处理、数据库操作等场景。

Model 信号

from django.db.models.signals import (
    pre_save,    # save() 调用前
    post_save,   # save() 调用后
    pre_delete,  # delete() 调用前
    post_delete, # delete() 调用后
    m2m_changed, # ManyToMany 字段变更时
    pre_init,    # Model.__init__() 调用前
    post_init,   # Model.__init__() 调用后
)

请求/响应信号

from django.core.signals import (
    request_started,   # 每个 HTTP 请求开始时
    request_finished,  # 每个 HTTP 请求结束时
    got_request_exception,  # 请求处理中发生异常时
)

数据库信号

from django.db.backends.signals import connection_created
# 数据库连接建立时触发

用户认证信号

from django.contrib.auth.signals import (
    user_logged_in,    # 用户登录成功
    user_logged_out,   # 用户登出
    user_login_failed, # 登录失败
)

注册接收者的两种方式

方式一：使用 @receiver 装饰器（推荐）

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model

User = get_user_model()

@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
    if created:
        # 新用户注册后，自动创建对应的 Profile
        UserProfile.objects.create(user=instance)

@receiver 的参数： - 第一个参数：要监听的信号（必填） - sender：只接收来自特定发送者的信号，不传则监听所有发送者

方式二：手动调用 connect()

def on_user_created(sender, instance, created, **kwargs):
    if created:
        UserProfile.objects.create(user=instance)

# 在 AppConfig.ready() 中连接
post_save.connect(on_user_created, sender=User)

两种方式等价，@receiver 只是 connect() 的语法糖，代码更简洁。

在哪里注册信号

这是新手最常踩的坑。如果在 models.py 或模块顶层直接写信号注册，可能出现信号被注册多次或信号未被加载的问题。

正确做法：在 App 的 AppConfig.ready() 方法中导入信号模块。

第一步：创建 signals.py

# myapp/signals.py
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from .models import Article, Tag

@receiver(post_save, sender=Article)
def article_saved(sender, instance, created, **kwargs):
    if created:
        print(f"新文章发布：{instance.title}")

@receiver(post_delete, sender=Article)
def article_deleted(sender, instance, **kwargs):
    # 文章删除后，清理孤立的标签
    Tag.objects.filter(article=None).delete()

第二步：在 AppConfig.ready() 中导入

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

class MyappConfig(AppConfig):
    name = 'myapp'
    default_auto_field = 'django.db.models.BigAutoField'

    def ready(self):
        import myapp.signals  # 导入触发注册，不需要使用返回值

第三步：确保 apps.py 被加载

# myapp/__init__.py
default_app_config = 'myapp.apps.MyappConfig'

或者在 settings.py 的 INSTALLED_APPS 中使用完整路径：

INSTALLED_APPS = [
    'myapp.apps.MyappConfig',  # 而不是 'myapp'
    ...
]

实战案例一：用户注册后自动发送欢迎邮件

# accounts/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.core.mail import send_mail
from django.contrib.auth import get_user_model

User = get_user_model()

@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if not created:
        return  # 只处理新建，忽略更新

    send_mail(
        subject='欢迎注册',
        message=f'Hi {instance.username}，欢迎加入！',
        from_email='[email protected]',
        recipient_list=[instance.email],
        fail_silently=True,  # 邮件失败不影响主流程
    )

注意：post_save 会在每次 save() 时触发，必须用 created 参数区分新建和更新。created=True 表示这是 INSERT，False 表示 UPDATE。

实战案例二：文章保存后自动清理缓存

博客系统中，文章更新后需要清除对应的页面缓存，否则用户还会看到旧内容：

# blog/signals.py
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from .models import Article

@receiver(post_save, sender=Article)
@receiver(post_delete, sender=Article)
def clear_article_cache(sender, instance, **kwargs):
    # 清除文章详情页缓存
    cache_key = f'article_detail_{instance.pk}'
    cache.delete(cache_key)

    # 清除文章列表页缓存（分类页、首页等）
    cache.delete('article_list_page_1')
    cache.delete(f'category_{instance.category_id}_articles')

    # 也可以用前缀批量清除（需要 Redis 后端）
    # cache.delete_pattern('article_*')

同一个函数可以用多个 @receiver 装饰器注册到不同信号。

实战案例三：M2M 变更时同步统计数据

文章的标签是多对多关系，标签变更时需要更新标签的文章计数：

# blog/signals.py
from django.db.models.signals import m2m_changed
from django.dispatch import receiver
from .models import Article, Tag

@receiver(m2m_changed, sender=Article.tags.through)
def update_tag_count(sender, instance, action, pk_set, **kwargs):
    # action 取值：pre_add / post_add / pre_remove / post_remove / pre_clear / post_clear
    if action not in ('post_add', 'post_remove', 'post_clear'):
        return  # 只处理操作完成后的事件

    if pk_set:
        # 更新受影响的标签计数
        tags = Tag.objects.filter(pk__in=pk_set)
    else:
        # post_clear 时 pk_set 为 None，更新所有标签
        tags = Tag.objects.all()

    for tag in tags:
        tag.article_count = tag.article_set.count()
        tag.save(update_fields=['article_count'])

m2m_changed 信号的 sender 不是 Model 本身，而是中间表（Article.tags.through），这是与其他信号的主要区别。

实战案例四：登录行为审计

# accounts/signals.py
from django.contrib.auth.signals import user_logged_in, user_login_failed, user_logged_out
from django.dispatch import receiver

@receiver(user_logged_in)
def log_login(sender, request, user, **kwargs):
    ip = request.META.get('REMOTE_ADDR', '')
    print(f"[登录成功] 用户: {user.username}, IP: {ip}")
    # 写入审计日志或数据库

@receiver(user_login_failed)
def log_login_failed(sender, credentials, request, **kwargs):
    ip = request.META.get('REMOTE_ADDR', '')
    username = credentials.get('username', '')
    print(f"[登录失败] 用户名: {username}, IP: {ip}")
    # 可以在这里实现登录失败次数限制

@receiver(user_logged_out)
def log_logout(sender, request, user, **kwargs):
    if user:
        print(f"[登出] 用户: {user.username}")

自定义信号

除了使用内置信号，也可以定义自己的业务信号：

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

# 定义信号，声明会传递的参数
article_published = Signal()
payment_completed = Signal()

在业务代码中发送信号：

# 在视图或 Model 方法中
from myapp.signals import article_published

def publish_article(article):
    article.status = 'p'
    article.save()
    # 发送自定义信号
    article_published.send(
        sender=article.__class__,
        article=article,
        publisher=request.user,
    )

注册接收者：

from myapp.signals import article_published

@receiver(article_published)
def notify_subscribers(sender, article, publisher, **kwargs):
    # 通知订阅了该作者的用户
    subscribers = article.author.followers.all()
    for user in subscribers:
        Notification.objects.create(
            recipient=user,
            message=f'{publisher.username} 发布了新文章：{article.title}',
        )

断开信号连接

有时需要临时禁用信号，比如在批量导入数据时不想触发副作用：

from django.db.models.signals import post_save

# 断开连接
post_save.disconnect(send_welcome_email, sender=User)

# 批量操作...
User.objects.bulk_create(users)

# 重新连接
post_save.connect(send_welcome_email, sender=User)

或者用上下文管理器封装，更优雅：

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

@contextmanager
def disable_signals(signal, receiver, sender):
    signal.disconnect(receiver, sender=sender)
    try:
        yield
    finally:
        signal.connect(receiver, sender=sender)

# 使用
with disable_signals(post_save, send_welcome_email, User):
    User.objects.bulk_create(users)

信号 vs 直接调用：如何选择

信号并不总是最佳方案，滥用信号会让代码变得难以追踪（"这个逻辑为什么被触发了？"）。

适合使用信号的场景： - 两个 App 之间需要解耦，不希望相互 import（如 blog app 触发 notification app 的逻辑） - 框架层的扩展点（如第三方库监听你的 Model 变化） - 审计日志、监控埋点等横切关注点

不适合使用信号的场景： - 同一个 App 内部的逻辑，直接调用更清晰 - 需要信号接收者的返回值（信号是单向通知，无法获取返回值） - 对性能敏感的热路径（信号有一定的调度开销） - 逻辑顺序很重要的地方（多个接收者的执行顺序不完全可控）

一个简单的判断标准：如果你需要在接收者里 import 发送者所在的模块，说明耦合度已经很高，信号的解耦价值不大，直接调用反而更清晰。

调试信号

当信号表现不符合预期时，可以用以下方式排查：

# 查看某个信号当前注册的所有接收者
from django.db.models.signals import post_save
from myapp.models import Article

receivers = post_save.receivers
print(f"post_save 共有 {len(receivers)} 个接收者")

# 临时打印所有触发的信号（调试用）
from django.db.models.signals import post_save

def debug_receiver(sender, **kwargs):
    print(f"post_save triggered: sender={sender}, kwargs={kwargs}")

post_save.connect(debug_receiver)

也可以使用 Django Debug Toolbar 或日志来追踪信号的触发情况。

总结

Django 信号机制的核心价值是解耦——让不同 App 的逻辑可以互相响应，而无需直接依赖。

几个关键实践：

统一在 AppConfig.ready() 中注册，避免重复注册和加载顺序问题
post_save 必须判断 created 参数，区分新建和更新
m2m_changed 的 sender 是中间表，不是 Model 本身
批量操作时考虑临时断开信号，避免性能问题
不要滥用信号，同一 App 内的逻辑直接调用更清晰

信号是 Django 架构设计中的重要工具，用在对的地方可以让代码结构清晰、职责分明；用错了则会让代码像"魔法"一样难以理解和调试。

本文由 liangliangyy 原创，转载请注明出处。