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

一、实时通信的核心：WebSocket 协议

WebSocket 介绍

在 Web 开发中，实时通信场景（如在线聊天、实时数据监控、协作工具等）越来越常见。传统的 HTTP 协议基于 "请求 - 响应" 模式，无法满足实时双向通信的需求。
WebSocket 是一种在单个 TCP 连接上提供全双工通信的网络协议，它打破了 HTTP 的单向请求限制，让服务器和客户端可以随时向对方发送数据。
WebSocket 的工作流程

• 握手阶段：通信始于 HTTP 请求，客户端（通常是浏览器）发送包含Upgrade: websocket字段的请求，申请升级WebSocket 协议
  
• 协议切换：服务器同意后返回101 Switching Protocols响应，连接从 HTTP 切换为 WebSocket
  
• 数据传输：此后通过帧（Frame）格式传输数据，支持文本帧、二进制帧和控制帧（关闭、ping/pong）。文本帧用于传输文本数据，二进制帧用于传输二进制数据，控制帧用于维护连接状态等操作。
  

WebSocket 的优势

相比传统的 HTTP 轮询（Polling）或长轮询（Long Polling），WebSocket 具有明显优势：

• 低延迟：建立一次连接后持续复用，避免重复握手开销，减少了通信的延迟
  
• 全双工：服务器可主动推送数据，无需客户端频繁请求
  
• 高效率：帧结构简洁，头部信息远小于 HTTP 头，减少传输开销，提高数据传输效率
  

WebSocket 的应用场景

• 实时聊天应用
  
• 实时协作工具
  
• 金融实时数据监控
  
• 在线游戏与互动应用
  
• 物联网（IoT）设备控制
  
• 实时通知系统
  

二、异步 Web 的标准：ASGI

为什么需要 ASGI？

要在 Django 中使用 WebSocket，需要依赖 ASGI（Asynchronous Server Gateway Interface）——Python 异步 Web 应用的标准接口。

• 传统的 WSGI 接口仅支持同步操作，无法处理 WebSocket 等异步场景
  
• ASGI 基于 asyncio 实现，支持异步 I/O，能高效处理并发连接
  
• 兼容 WSGI 应用，可通过适配器运行现有同步应用
  

ASGI 的核心特性

• 原生支持 WebSocket 和 HTTP/2 等实时协议
  
• 异步非阻塞处理，提升高并发场景下的性能
  
• 标准化接口，让框架（如 Django、FastAPI）和服务器（如 Uvicorn）可无缝协作
  

参考资料：ASGI 项目文档
三、ASGI 服务器：Uvicorn

Uvicorn介绍

Uvicorn 是一款高性能的 ASGI 服务器，专为异步 Web 应用设计，是运行 Django 实时应用的理想选择。
安装uvicorn

1. # standard 是指包含WebSocket支持的完整版
2. pip install "uvicorn[standard]"
4. # 查看版本
5. uvicorn --version

复制代码

运行 Django 项目

Django 项目默认包含asgi.py入口文件，通过 Uvicorn 启动。进入项目根目录（包含manage.py的目录），运行下面命令

1. # 开发环境（自动重载）
2. uvicorn mysite.asgi:application --reload
4. # 或指定地址和端口
5. uvicorn mysite.asgi:application --host 0.0.0.0 --port 8000 --reload

复制代码

参考资料：uvicorn官方文档
四、Django Channels：让 Django 支持 WebSocket

Django Channels介绍

Django 本身不直接支持 WebSocket，需要通过Channels扩展实现。Channels 为 Django 添加了异步处理能力，使其能处理 WebSocket、HTTP2 等协议。
Django Channels的核心组件

• Consumers（消费者）：类似视图（View），处理 WebSocket 连接、消息收发
  
• Routing（路由）：将 WebSocket URL 映射到对应的消费者
  
• Channel Layers（通道层）：跨进程 / 服务器的消息代理，支持 Redis 等后端
  
• Protocols：支持 HTTP、WebSocket 等协议
  

安装与配置

安装

1. pip install channels
2. pip install channels_redis

复制代码

安装Redis（过程略）
配置Django项目的mysite\mysite\settings.py

1. INSTALLED_APPS = [
2.     # ...其他应用
3.     "channels",  # 添加Channels
4. ]
6. # 指定ASGI应用入口
7. ASGI_APPLICATION = "mysite.asgi.application"
9. # 配置Redis通道层
10. CHANNEL_LAYERS = {
11.     "default": {
12.         "BACKEND": "channels_redis.core.RedisChannelLayer",
13.         "CONFIG": {
14.             "hosts": ["redis://127.0.0.1:6379/3"],  # Redis地址
15.         },
16.     },
17. }

复制代码

点击查看完整代码
配置Django项目的mysite\mysite\asgi.py

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. os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
7. django_application = get_asgi_application()
9. # 延迟导入WebSocket路由（避免循环导入）
10. def get_websocket_application():
11.     from myapp.websocket.routing import websocket_urlpatterns
12.     return AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
14. # 协议路由：区分HTTP和WebSocket请求
15. application = ProtocolTypeRouter({
16.     "http": django_application,
17.     "websocket": get_websocket_application()
18. })

复制代码

点击查看完整代码
消费者示例

消费者（Consumer）是处理 WebSocket 逻辑的核心，类似 Django 的视图，支持同步和异步两种模式。
消费者类型

• AsyncWebsocketConsumer：异步处理（推荐），适合高并发场景
  
• SyncWebsocketConsumer：同步处理，适合简单逻辑或需集成同步代码
  
• JsonWebsocketConsumer：专门处理 JSON 格式消息，自动解析 / 序列化
  

异步消费者示例

• connect() ：当客户端发起 WebSocket 连接时调用。
  
• disconnect() ：当客户端关闭 WebSocket 连接时调用。
  
• receive() ：当接收到客户端发送的消息时调用。
  
• send() ：用于向客户端发送消息。
  

1. from channels.generic.websocket import AsyncWebsocketConsumer
2. import json
4. class ChatConsumer(AsyncWebsocketConsumer):
5.     # 连接建立时调用
6.     async def connect(self):
7.         # 从URL获取房间名
8.         self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
9.         self.room_group_name = f"chat_{self.room_name}"
10.         
11.         # 加入房间组
12.         await self.channel_layer.group_add(
13.             self.room_group_name,
14.             self.channel_name
15.         )
16.         # 接受连接
17.         await self.accept()
19.     # 连接关闭时调用
20.     async def disconnect(self, close_code):
21.         # 离开房间组
22.         await self.channel_layer.group_discard(
23.             self.room_group_name,
24.             self.channel_name
25.         )
27.     # 接收客户端消息
28.     async def receive(self, text_data):
29.         text_data_json = json.loads(text_data)
30.         message = text_data_json["message"]
31.         
32.         # 发送消息到房间组（广播给同组所有连接）
33.         await self.channel_layer.group_send(
34.             self.room_group_name,
35.             {"type": "chat.message", "message": message}
36.         )
38.     # 处理房间组消息（方法名对应group_send中的type字段）
39.     async def chat.message(self, event):
40.         message = event["message"]
41.         # 发送消息给客户端
42.         await self.send(text_data=json.dumps({"message": message}))

复制代码

路由配置示例

通过路由将 WebSocket URL 映射到消费者，类似 Django 的 URL 配置

1. # routing.py
2. from django.urls import re_path
3. from . import consumers
5. websocket_urlpatterns = [
6.     re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
7. ]

复制代码

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

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