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 AuthMiddlewareStackos.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
django_application = get_asgi_application()# 延迟导入WebSocket路由（避免循环导入）
def get_websocket_application():from myapp.websocket.routing import websocket_urlpatternsreturn 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 jsonclass 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 consumerswebsocket_urlpatterns = [re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
]

---

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