Django中如何无需刷新页面自动检测数据库新增用户条目

核心逻辑说明 #

Django的post_save信号可以直接监听User表的新增动作，符合你优先使用signal方案的要求。由于信号是服务端内部的事件钩子，无法直接把事件推送给前端浏览器，因此搭配WebSocket实现服务端到前端的主动推送，完全替代AJAX轮询，避免多线程环境下的查询性能问题。

具体实现步骤 #

• 第一步：安装WebSocket依赖
  Django 3.2原生支持ASGI协议，使用channels作为WebSocket实现，执行以下命令安装依赖：
  pip install channels channels-redis
  在settings.py中添加基础配置：

INSTALLED_APPS = [
    # 其他原有应用
    'channels',
]

# 指定ASGI路由入口
ASGI_APPLICATION = '你的项目名.asgi.application'

# 配置通道层，使用Redis作为中间存储
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

• 第二步：编写User表新增监听信号
  在用户相关应用下新建signals.py，监听用户创建事件：

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User # 如果是自定义用户模型，替换为你自己的模型类
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

@receiver(post_save, sender=User)
def notify_new_user_register(sender, instance, created, **kwargs):
    # 仅用户新增时触发推送
    if created:
        channel_layer = get_channel_layer()
        # 向所有订阅新用户通知的前端推送数据
        async_to_sync(channel_layer.group_send)(
            "new_user_notify_group",
            {
                "type": "push_new_user",
                "data": {
                    "id": instance.id,
                    "username": instance.username,
                    "phone": instance.phone, # 按需添加你需要返回的用户字段
                }
            }
        )

在对应应用的apps.py中注册信号：

from django.apps import AppConfig

class 你的应用名Config(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = '你的应用名'

    def ready(self):
        import 你的应用名.signals

• 第三步：编写WebSocket消费者
  处理前端的WebSocket连接请求，接收信号推送的消息并转发给前端：

from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_sync
import json

class NewUserConsumer(WebsocketConsumer):
    def connect(self):
        self.group_name = "new_user_notify_group"
        # 把当前连接加入广播组
        async_to_sync(self.channel_layer.group_add)(
            self.group_name,
            self.channel_name
        )
        self.accept()

    def disconnect(self, close_code):
        # 断开连接时移出广播组
        async_to_sync(self.channel_layer.group_discard)(
            self.group_name,
            self.channel_name
        )

    # 接收信号发来的新用户数据，推送给前端
    def push_new_user(self, event):
        self.send(text_data=json.dumps(event["data"]))

在项目根目录的asgi.py中配置WebSocket路由：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from 你的应用名.routing import websocket_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(websocket_urlpatterns)
    ),
})

在应用下新建routing.py定义WebSocket路由规则：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/new/user/$', consumers.NewUserConsumer.as_asgi()),
]

• 第四步：前端添加WebSocket连接逻辑
  不需要AJAX轮询，直接通过WebSocket接收服务端主动推送的新用户数据，更新页面：

// 建立WebSocket连接
const userNotifySocket = new WebSocket(`ws://${window.location.host}/ws/new/user/`);

// 接收新用户通知
userNotifySocket.onmessage = function (e) {
    const newUser = JSON.parse(e.data);
    // 此处编写你的页面更新逻辑，比如把新用户插入到用户列表顶部
    console.log('新注册用户：', newUser);
};

// 连接断开自动重连
userNotifySocket.onclose = function (e) {
    console.error('WebSocket连接断开，3秒后尝试重连');
    setTimeout(() => window.location.reload(), 3000);
};

注意事项 #

• 开发环境和生产环境都需要使用ASGI服务器启动项目（比如daphne、uvicorn），不要使用旧的WSGI服务器，否则WebSocket无法生效
• 如果不需要广播给所有前端，而是只推送给指定权限的管理员，可以在消费者的connect方法中增加权限校验逻辑，不符合权限的连接直接拒绝即可
• 生产环境的Redis建议配置密码和持久化，避免通道层数据丢失

内容的提问来源于stack exchange，提问作者azhar