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项目的mysite\mysite\settings.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项目的mysite\mysite\asgi.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()), ]

您正在阅读的是《