Django实时通信实战：WebSocket与ASGI全解析（下）

一、实战：构建实时聊天室

环境准备

下面将使用 Django Channels 构建一个多用户实时聊天室。Django Channels的介绍、安装与配置，参考上篇。

实现 WebSocket 消费者

创建mysitemyapp_infrawebsocketconsumers.py文件，这是处理 WebSocket 连接的核心。主要实现了如下方法：

• connect：处理新连接，验证用户token，将用户通道写入缓存，并加入默认组
• disconnect：处理连接断开，删除用户的缓存通道记录，将用户从所在的房间组中移除
• receive：处理接收到的消息。首先进行心跳检测（ping/pong），然后验证用户身份，解析两层 JSON 结构的消息内容。根据消息类型 demo-message-send 处理文本消息：若指定接收用户则单发，否则群发广播。
• broadcast_message：处理群发消息，从事件中提取payload数据并发送给所有连接的客户端
• single_message：处理单发消息，从事件中提取payload数据并发送给指定的客户端

"""WebScoket 消费者"""  import json import logging from django.core.cache import cache from django.conf import settings from channels.generic.websocket import AsyncWebsocketConsumer from rest_framework_simplejwt.tokens import RefreshToken from rest_framework_simplejwt.exceptions import TokenError  logger = logging.getLogger(__name__)   class InfraConsumer(AsyncWebsocketConsumer):     """基础设施-WebSocket 异步消费者"""      async def connect(self):         """当客户端发起 WebSocket 连接时调用"""         query_params = self.scope["query_string"].decode()         # 从查询参数获取token         token_param = [             param.split("=")             for param in query_params.split("&")             if param.startswith("token=")         ]         if not token_param or len(token_param[0]) != 2:             logger.error("缺少token参数")             await self.close(code=4001)  # 自定义错误码             return          token = token_param[0][1]         try:             # 验证Refresh Token有效性             refresh = RefreshToken(token)             user_id = refresh["user_id"]             self.scope["user"] = {"id": user_id, "token_type": "refresh"}              # 登录成功后，将用户通道写入缓存             cache.set(f"user_{user_id}_channel", self.channel_name)             # 加入默认组             self.room_group_name = settings.DEFAULT_GROUP_NAME             await self.channel_layer.group_add(self.room_group_name, self.channel_name)             # 接受客户端的WebSocket连接请求             await self.accept()         except TokenError as e:             logger.error("无效token")             await self.close(code=4003)         except Exception as e:             logger.error(str(e))             await self.close(code=4000)      async def disconnect(self, close_code):         """当客户端断开 WebSocket 连接时调用"""         # 获取当前用户         user = self.scope.get("user")         if user:             user_id = user.get("id")             # 移除用户通道             cache.delete(f"user_{user_id}_channel")             # 将用户从组中移除             room_group_name = getattr(self, "room_group_name", None)             if room_group_name:                 await self.channel_layer.group_discard(                     room_group_name, self.channel_name                 )      async def receive(self, text_data=None, bytes_data=None):         """当接收到客户端发送的消息时调用"""         # 心跳检测         if text_data == "ping":             await self.send(text_data="pong")             return          user = self.scope.get("user")         if not user:             logger.warning(f"匿名用户访问拒绝：{text_data}")             return         logger.info(f"收到用户 {user.get('id')} 发送消息：{text_data}")          # 因为消息采用了两次JSON封装，这里进行两次JSON解析         try:             outer_payload = json.loads(text_data)  # 外层JSON解释             message_type = outer_payload.get("type")             raw_content = outer_payload.get("content", "{}")             inner_content = json.loads(raw_content)  # 内层JSON解释         except json.JSONDecodeError:             logger.error("内容解析失败")             return          # 处理业务消息         if message_type == "demo-message-send":             message_text = inner_content.get("text", "").strip()             if not message_text:                 logger.warning("消息内容不能为空")                 return              # 构建响应字典             content = {                 "fromUserId": user.get("id"),                 "text": message_text,                 "single": False,  # 默认群发             }              # 判断接收对象             target_user_id = inner_content.get("toUserId")             if target_user_id not in [None, "", 0]:                 # 单发                 await self._send_single_message(target_user_id, content)             else:                 # 群发广播                 await self._send_broadcast_message(content)      def _build_response(self, content):         """构建标准化的响应消息，进行两次JSON封装"""         # 第一次封装：添加消息类型         inner_message = {             "type": "demo-message-receive",             "content": json.dumps(content),         }         # 第二次封装：整体序列化         return json.dumps(inner_message)      async def _send_single_message(self, target_user_id, content):         """向指定用户发送单条消息"""         # 获取用户通道         target_channel = cache.get(f"user_{target_user_id}_channel")         if not target_channel:             return False          # 构建并发送消息         message = self._build_response(content)         await self.channel_layer.send(             target_channel,             {                 "type": "single.message",                 "payload": message,             },         )         return True      async def _send_broadcast_message(self, content):         """向房间内所有用户广播消息"""         message = self._build_response(content)         await self.channel_layer.group_send(             self.room_group_name,             {                 "type": "broadcast.message",                 "payload": message,             },         )      async def broadcast_message(self, event):         """处理群发消息"""         payload = event["payload"]         await self.send(text_data=payload)      async def single_message(self, event):         """处理单发消息"""         payload = event["payload"]         await self.send(text_data=payload) 

点击查看完整代码

配置 WebSocket 路由

创建mysitemyapp_infrawebsocketrouting.py文件，定义 WebSocket 的 URL 路由

"""WebSocket 路由配置"""  from django.urls import re_path, path from .consumers import InfraConsumer  websocket_urlpatterns = [     # 调用as_asgi()类方法来获取一个 ASGI 应用程序     re_path(r"^infra/ws/?$", InfraConsumer.as_asgi()), ] 

然后在项目的mysitemysiteasgi.py配置中添加 ASGI 路由

import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack  # 设置环境变量并初始化Django应用 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings") django_application = get_asgi_application()   def get_websocket_application():     """延迟导入WebSocket路由（避免循环导入）"""     from myapp_infra.websocket.routing import websocket_urlpatterns      return AuthMiddlewareStack(URLRouter(websocket_urlpatterns))   # 协议路由：区分HTTP和WebSocket请求 application = ProtocolTypeRouter(     {         "http": django_application,         "websocket": get_websocket_application(),     } ) 

点击查看完整代码

创建聊天界面模板

以Vue3界面为例，代码实现了一个 WebSocket 客户端界面，功能包括：

• 连接控制：显示连接状态、开启/关闭 WebSocket 连接。
• 消息发送：输入消息内容并选择接收人（单发或群发），点击发送按钮通过 WebSocket 发送消息。
• 消息接收与展示：监听 WebSocket 返回的数据，解析不同类型的消息（如单发、群发、系统通知）并在右侧列表中倒序展示。
• 用户列表获取：页面加载时获取用户列表用于选择消息接收人。

点击查看完整代码

实现效果

使用 ASGI 运行项目

# 开发环境启动 uvicorn mysite.asgi:application --reload 

使用不同的浏览器，分别登录不同的用户，实现实时互发消息。

二、生产环境部署

Nginx 配置

使用 Nginx 作为反向代理，添加以下配置来处理 WebSocket 连接。这段配置告诉 Nginx 如何正确处理 WebSocket 升级请求。

location /ws/ {     proxy_pass http://backend;     proxy_http_version 1.1;     proxy_set_header Upgrade $http_upgrade;     proxy_set_header Connection "upgrade"; } 

使用 Gunicorn+Uvicorn 部署

Gunicorn（全称 Green Unicorn）是一个用于 UNIX 的 高性能 Python WSGI HTTP 服务器。Gunicorn只能用于 Linux 系统，Windows上无法使用。

安装

pip install gunicorn  # 查看版本 gunicorn -v gunicorn -h 

Gunicorn 作为进程管理器，配合 Uvicorn 工作进程处理 ASGI 应用

• -w 工作进程的数量。默认启动 1 个 Worker 进程
• -k 要运行的工作进程类型。
• -b 指定要绑定的服务器套接字。

# 基本启动命令 gunicorn -b 0.0.0.0:8000 -k uvicorn.workers.UvicornWorker mysite.asgi:application 

优化 Worker 数量

Gunicorn Worker 数量的设置对性能影响很大，推荐的公式是：(CPU核心数 × 2) + 1。我们可以编写一个启动脚本来自动计算最佳 Worker 数量

#!/bin/bash # run_gunicorn.sh  # 计算最佳 Worker 数 CORES=$(nproc) WORKERS=$((CORES * 2 + 1))  # 限制最大Worker数（避免内存不足） MAX_WORKERS=8 if [ $WORKERS -gt $MAX_WORKERS ]; then   WORKERS=$MAX_WORKERS fi  # 启动命令 gunicorn -b 0.0.0.0:8000    --workers $WORKERS    --max-requests 1000          # 预防内存泄漏   --timeout 120                # 超时控制   --keep-alive 5               # Keep-Alive   -k uvicorn.workers.UvicornWorker    mysite.asgi:application 

添加执行权限并运行

chmod +x run_gunicorn.sh ./run_gunicorn.sh 

---

您正在阅读的是《Django从入门到实战》专栏！关注不迷路~