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

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

运行 Django 项目

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

# 开发环境（自动重载） uvicorn mysite.asgi:application --reload  # 或指定地址和端口 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 等协议

安装与配置

安装

pip install channels pip install channels_redis 

安装Redis（过程略）

配置Django项目的mysitemysitesettings.py

INSTALLED_APPS = [     # ...其他应用     "channels",  # 添加Channels ]  # 指定ASGI应用入口 ASGI_APPLICATION = "mysite.asgi.application"  # 配置Redis通道层 CHANNEL_LAYERS = {     "default": {         "BACKEND": "channels_redis.core.RedisChannelLayer",         "CONFIG": {             "hosts": ["redis://127.0.0.1:6379/3"],  # Redis地址         },     }, } 

点击查看完整代码

配置Django项目的mysitemysiteasgi.py

import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack  os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings") django_application = get_asgi_application()  # 延迟导入WebSocket路由（避免循环导入） def get_websocket_application():     from myapp.websocket.routing import websocket_urlpatterns     return AuthMiddlewareStack(URLRouter(websocket_urlpatterns))  # 协议路由：区分HTTP和WebSocket请求 application = ProtocolTypeRouter({     "http": django_application,     "websocket": get_websocket_application() }) 

点击查看完整代码

消费者示例

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

消费者类型

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

异步消费者示例

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

from channels.generic.websocket import AsyncWebsocketConsumer import json  class ChatConsumer(AsyncWebsocketConsumer):     # 连接建立时调用     async def connect(self):         # 从URL获取房间名         self.room_name = self.scope["url_route"]["kwargs"]["room_name"]         self.room_group_name = f"chat_{self.room_name}"                  # 加入房间组         await self.channel_layer.group_add(             self.room_group_name,             self.channel_name         )         # 接受连接         await self.accept()      # 连接关闭时调用     async def disconnect(self, close_code):         # 离开房间组         await self.channel_layer.group_discard(             self.room_group_name,             self.channel_name         )      # 接收客户端消息     async def receive(self, text_data):         text_data_json = json.loads(text_data)         message = text_data_json["message"]                  # 发送消息到房间组（广播给同组所有连接）         await self.channel_layer.group_send(             self.room_group_name,             {"type": "chat.message", "message": message}         )      # 处理房间组消息（方法名对应group_send中的type字段）     async def chat.message(self, event):         message = event["message"]         # 发送消息给客户端         await self.send(text_data=json.dumps({"message": message})) 

路由配置示例

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

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

---

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