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

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

环境准备

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

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

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

1. """WebScoket 消费者"""
3. import json
4. import logging
5. from django.core.cache import cache
6. from django.conf import settings
7. from channels.generic.websocket import AsyncWebsocketConsumer
8. from rest_framework_simplejwt.tokens import RefreshToken
9. from rest_framework_simplejwt.exceptions import TokenError
11. logger = logging.getLogger(__name__)
14. class InfraConsumer(AsyncWebsocketConsumer):
15.     """基础设施-WebSocket 异步消费者"""
17.     async def connect(self):
18.         """当客户端发起 WebSocket 连接时调用"""
19.         query_params = self.scope["query_string"].decode()
20.         # 从查询参数获取token
21.         token_param = [
22.             param.split("=")
23.             for param in query_params.split("&")
24.             if param.startswith("token=")
25.         ]
26.         if not token_param or len(token_param[0]) != 2:
27.             logger.error("缺少token参数")
28.             await self.close(code=4001)  # 自定义错误码
29.             return
31.         token = token_param[0][1]
32.         try:
33.             # 验证Refresh Token有效性
34.             refresh = RefreshToken(token)
35.             user_id = refresh["user_id"]
36.             self.scope["user"] = {"id": user_id, "token_type": "refresh"}
38.             # 登录成功后，将用户通道写入缓存
39.             cache.set(f"user_{user_id}_channel", self.channel_name)
40.             # 加入默认组
41.             self.room_group_name = settings.DEFAULT_GROUP_NAME
42.             await self.channel_layer.group_add(self.room_group_name, self.channel_name)
43.             # 接受客户端的WebSocket连接请求
44.             await self.accept()
45.         except TokenError as e:
46.             logger.error("无效token")
47.             await self.close(code=4003)
48.         except Exception as e:
49.             logger.error(str(e))
50.             await self.close(code=4000)
52.     async def disconnect(self, close_code):
53.         """当客户端断开 WebSocket 连接时调用"""
54.         # 获取当前用户
55.         user = self.scope.get("user")
56.         if user:
57.             user_id = user.get("id")
58.             # 移除用户通道
59.             cache.delete(f"user_{user_id}_channel")
60.             # 将用户从组中移除
61.             room_group_name = getattr(self, "room_group_name", None)
62.             if room_group_name:
63.                 await self.channel_layer.group_discard(
64.                     room_group_name, self.channel_name
65.                 )
67.     async def receive(self, text_data=None, bytes_data=None):
68.         """当接收到客户端发送的消息时调用"""
69.         # 心跳检测
70.         if text_data == "ping":
71.             await self.send(text_data="pong")
72.             return
74.         user = self.scope.get("user")
75.         if not user:
76.             logger.warning(f"匿名用户访问拒绝：{text_data}")
77.             return
78.         logger.info(f"收到用户 {user.get('id')} 发送消息：{text_data}")
80.         # 因为消息采用了两次JSON封装，这里进行两次JSON解析
81.         try:
82.             outer_payload = json.loads(text_data)  # 外层JSON解释
83.             message_type = outer_payload.get("type")
84.             raw_content = outer_payload.get("content", "{}")
85.             inner_content = json.loads(raw_content)  # 内层JSON解释
86.         except json.JSONDecodeError:
87.             logger.error("内容解析失败")
88.             return
90.         # 处理业务消息
91.         if message_type == "demo-message-send":
92.             message_text = inner_content.get("text", "").strip()
93.             if not message_text:
94.                 logger.warning("消息内容不能为空")
95.                 return
97.             # 构建响应字典
98.             content = {
99.                 "fromUserId": user.get("id"),
100.                 "text": message_text,
101.                 "single": False,  # 默认群发
102.             }
104.             # 判断接收对象
105.             target_user_id = inner_content.get("toUserId")
106.             if target_user_id not in [None, "", 0]:
107.                 # 单发
108.                 await self._send_single_message(target_user_id, content)
109.             else:
110.                 # 群发广播
111.                 await self._send_broadcast_message(content)
113.     def _build_response(self, content):
114.         """构建标准化的响应消息，进行两次JSON封装"""
115.         # 第一次封装：添加消息类型
116.         inner_message = {
117.             "type": "demo-message-receive",
118.             "content": json.dumps(content),
119.         }
120.         # 第二次封装：整体序列化
121.         return json.dumps(inner_message)
123.     async def _send_single_message(self, target_user_id, content):
124.         """向指定用户发送单条消息"""
125.         # 获取用户通道
126.         target_channel = cache.get(f"user_{target_user_id}_channel")
127.         if not target_channel:
128.             return False
130.         # 构建并发送消息
131.         message = self._build_response(content)
132.         await self.channel_layer.send(
133.             target_channel,
134.             {
135.                 "type": "single.message",
136.                 "payload": message,
137.             },
138.         )
139.         return True
141.     async def _send_broadcast_message(self, content):
142.         """向房间内所有用户广播消息"""
143.         message = self._build_response(content)
144.         await self.channel_layer.group_send(
145.             self.room_group_name,
146.             {
147.                 "type": "broadcast.message",
148.                 "payload": message,
149.             },
150.         )
152.     async def broadcast_message(self, event):
153.         """处理群发消息"""
154.         payload = event["payload"]
155.         await self.send(text_data=payload)
157.     async def single_message(self, event):
158.         """处理单发消息"""
159.         payload = event["payload"]
160.         await self.send(text_data=payload)

复制代码

点击查看完整代码
配置 WebSocket 路由

创建mysite\myapp_infra\websocket\routing.py文件，定义 WebSocket 的 URL 路由

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

复制代码

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

1. import os
2. from django.core.asgi import get_asgi_application
3. from channels.routing import ProtocolTypeRouter, URLRouter
4. from channels.auth import AuthMiddlewareStack
6. # 设置环境变量并初始化Django应用
7. os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
8. django_application = get_asgi_application()
11. def get_websocket_application():
12.     """延迟导入WebSocket路由（避免循环导入）"""
13.     from myapp_infra.websocket.routing import websocket_urlpatterns
15.     return AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
18. # 协议路由：区分HTTP和WebSocket请求
19. application = ProtocolTypeRouter(
20.     {
21.         "http": django_application,
22.         "websocket": get_websocket_application(),
23.     }
24. )

复制代码

点击查看完整代码
创建聊天界面模板

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

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

点击查看完整代码
实现效果

使用 ASGI 运行项目

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

复制代码

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

二、生产环境部署

Nginx 配置

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

1. location /ws/ {
2.     proxy_pass http://backend;
3.     proxy_http_version 1.1;
4.     proxy_set_header Upgrade $http_upgrade;
5.     proxy_set_header Connection "upgrade";
6. }

复制代码

使用 Gunicorn+Uvicorn 部署

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

1. pip install gunicorn
3. # 查看版本
4. gunicorn -v
5. gunicorn -h

复制代码

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

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

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

复制代码

优化 Worker 数量

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

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

复制代码

添加执行权限并运行

1. chmod +x run_gunicorn.sh
2. ./run_gunicorn.sh

复制代码

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

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！