如何为自研简易信使实现无刷新消息通知功能？

1. 环境配置 #

首先安装必要依赖：

pip install channels channels-redis

修改settings.py，添加Channels相关配置：

INSTALLED_APPS = [
    # 其他已安装应用
    'channels',
]

# ASGI配置
ASGI_APPLICATION = '你的项目名.asgi.application'

# 通道层配置（用Redis作为后端）
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

2. 创建WebSocket消费者 #

在你的信使应用下新建consumers.py，实现用户连接分组和消息推送逻辑：

from channels.generic.websocket import AsyncWebsocketConsumer
import json

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # 获取当前用户ID，创建专属分组
        self.user_id = self.scope['user'].id
        self.group_name = f'user_{self.user_id}'

        # 加入分组
        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        # 离开分组
        await self.channel_layer.group_discard(
            self.group_name,
            self.channel_name
        )

    # 定义接收消息的方法，用于接收信号发送的通知
    async def send_notification(self, event):
        message = event['message']
        # 向前端发送未读消息计数
        await self.send(text_data=json.dumps({
            'unread_count': message['unread_count']
        }))

3. 配置WebSocket路由 #

在项目根目录新建routing.py：

from django.urls import re_path
from .consumers import NotificationConsumer

websocket_urlpatterns = [
    re_path(r'ws/notifications/$', NotificationConsumer.as_asgi()),
]

修改asgi.py，整合Channels路由：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from . import routing

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

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

4. 用Django Signals触发实时通知 #

在你的消息模型所在的models.py中，添加post_save信号：

from django.db import models
from django.db.models.signals import post_save
from django.dispatch import receiver
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
from django.contrib.auth import get_user_model

User = get_user_model()

class Message(models.Model):
    sender = models.ForeignKey(User, on_delete=models.CASCADE, related_name='sent_messages')
    receiver = models.ForeignKey(User, on_delete=models.CASCADE, related_name='received_messages')
    content = models.TextField()
    is_read = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

@receiver(post_save, sender=Message)
def send_message_notification(sender, instance, created, **kwargs):
    if created:
        # 计算接收者的未读消息总数
        unread_count = instance.receiver.received_messages.filter(is_read=False).count()
        channel_layer = get_channel_layer()
        # 向接收者的分组发送通知
        async_to_sync(channel_layer.group_send)(
            f'user_{instance.receiver.id}',
            {
                'type': 'send_notification',
                'message': {
                    'unread_count': unread_count
                }
            }
        )

5. 前端实现WebSocket连接和通知展示 #

在页面模板中添加WebSocket连接代码，实时更新未读图标：

<!-- 未读通知图标 -->
<div id="notification-icon">
    <span id="unread-count" style="display:none; background: red; color: white; border-radius: 50%; padding: 2px 6px; font-size: 12px;">0</span>
</div>

<script>
// 建立WebSocket连接
const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
const wsUrl = `${wsProtocol}//${window.location.host}/ws/notifications/`;
const socket = new WebSocket(wsUrl);

socket.onmessage = function(e) {
    const data = JSON.parse(e.data);
    const unreadCount = data.unread_count;
    const countElement = document.getElementById('unread-count');
    
    if (unreadCount > 0) {
        countElement.textContent = unreadCount;
        countElement.style.display = 'inline';
    } else {
        countElement.style.display = 'none';
    }
};

socket.onclose = function(e) {
    console.error('WebSocket连接关闭，尝试重新连接...');
    setTimeout(() => window.location.reload(), 5000);
};
</script>

关键注意事项 #

• 确保Redis服务正在运行，否则通道层无法正常工作
• 消费者使用AsyncWebsocketConsumer实现异步处理，性能更优
• 信号中使用async_to_sync将异步的通道层调用转为同步，适配Django信号的同步触发机制
• 前端需处理WebSocket断开重连逻辑，避免用户长时间停留后失去实时通知能力

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