网站导航
> 技术文档 > Django实时通信实战：WebSocket与ASGI全解析（上）_uvicorn asgi websocket demo 心跳包 django5.1

Django实时通信实战：WebSocket与ASGI全解析（上）_uvicorn asgi websocket demo 心跳包 django5.1

网友: 技术文档 2025-07-26 20:41:49

文章目录

    • 一、实时通信的核心：WebSocket 协议
      • WebSocket 介绍
      • WebSocket 的工作流程
      • WebSocket 的优势
      • WebSocket 的应用场景
    • 二、异步 Web 的标准：ASGI
      • 为什么需要 ASGI？
      • ASGI 的核心特性
    • 三、ASGI 服务器：Uvicorn
      • Uvicorn介绍
      • 安装uvicorn
      • 运行 Django 项目
    • 四、Django Channels：让 Django 支持 WebSocket
      • Django Channels介绍
      • 安装与配置
      • 消费者示例
      • 路由配置示例

一、实时通信的核心：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 channelspip 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 osfrom django.core.asgi import get_asgi_applicationfrom channels.routing import ProtocolTypeRouter, URLRouterfrom 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_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 AsyncWebsocketConsumerimport 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.pyfrom django.urls import re_pathfrom . import consumerswebsocket_urlpatterns = [ re_path(r\"ws/chat/(?P\\w+)/$\", consumers.ChatConsumer.as_asgi()),]

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

网络标签：

相关问题