如何使用Django Channels实现聊天会话展示及实时更新？

首先明确结论：完全不需要为每个会话单独创建WebSocket连接，单用户单连接的方案足够覆盖你的需求，还能大幅降低服务端资源开销和连接管理复杂度。

• 单用户同时建立多连接会成倍占用服务端连接池资源，用户会话量大的情况下极易触发连接数上限
• 多连接下的消息同步、断连重连逻辑会复杂很多，容易出现会话更新不同步的问题
• 单个连接完全可以通过不同的消息类型区分全量会话拉取、单会话更新、新消息通知等不同场景

1. 自定义会话管理Consumer #

专门用一个Consumer处理用户会话列表相关的全量拉取和实时更新请求，可与具体的聊天消息Consumer分开部署实现解耦，示例代码如下：

from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async
from channels.layers import get_channel_layer
import json
from .models import Conversation

class UserSessionConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user = self.scope["user"]
        # 未认证用户直接拒绝连接
        if not self.user.is_authenticated:
            await self.close()
            return
        # 定义用户专属的会话更新分组名
        self.session_group_name = f"user_{self.user.id}_session_updates"
        # 当前连接加入分组
        await self.channel_layer.group_add(
            self.session_group_name,
            self.channel_name
        )
        await self.accept()
        # 连接建立后直接推送全量活跃会话数据，无需前端额外发起请求
        active_conversations = await self.get_active_conversations()
        await self.send(text_data=json.dumps({
            "type": "full_session_list",
            "data": active_conversations
        }))
    
    @database_sync_to_async
    def get_active_conversations(self):
        # 查询当前用户的所有活跃会话，按需返回前端需要的字段
        conversations = Conversation.objects.filter(
            participants=self.user,
            is_active=True
        ).order_by("-last_message_at")
        return [
            {
                "id": conv.id,
                "display_name": conv.get_display_name(self.user),
                "last_message": conv.last_message_content,
                "last_message_at": conv.last_message_at.isoformat(),
                "unread_count": conv.get_unread_count(self.user)
            } for conv in conversations
        ]
    
    # 处理分组内推送的会话更新事件
    async def session_update(self, event):
        await self.send(text_data=json.dumps({
            "type": "single_session_update",
            "data": event["data"]
        }))
    
    async def disconnect(self, code):
        # 断开连接时退出分组
        await self.channel_layer.group_discard(
            self.session_group_name,
            self.channel_name
        )

2. 通道层分组规则 #

• 每个已认证用户对应唯一的个人会话更新分组，命名规则统一为user_{user_id}_session_updates
• 所有和该用户相关的会话变更（新消息、未读数更新、会话归档/激活）都统一推送到这个分组，用户的单个连接就能收到所有会话的更新通知

3. 新消息触发会话更新逻辑 #

当用户在任意会话发送/收到新消息时，按以下流程处理：

1. 保存消息到数据库，同时更新对应会话的last_message_content、last_message_at、未读数字段
2. 分别给会话的所有参与方的个人会话更新分组推送该会话的最新数据，示例触发代码：

# 新消息保存成功后调用该方法推送更新
async def push_session_update(user_id, updated_session_data):
    channel_layer = get_channel_layer()
    await channel_layer.group_send(
        f"user_{user_id}_session_updates",
        {
            "type": "session_update",
            "data": updated_session_data
        }
    )

3. 前端收到single_session_update类型的消息后，直接替换列表中对应会话的内容即可，不需要重新拉取全量列表

• 会话页面加载完成后首先建立和UserSessionConsumer的WebSocket连接
• 连接建立成功后会收到服务端主动推送的full_session_list类型消息，直接渲染全量会话列表
• 后续收到single_session_update类型的增量更新消息，匹配会话ID后更新对应条目的最新消息、时间、未读数即可

内容的提问来源于stack exchange，提问作者Samy van Stokkum