小王子1024

專注Django實戰案例分享

Django實時通信實戰：WebSocket與ASGI全解析（上）

WebSocket 是一種全雙工通信協議，支持實時數據傳輸，適用于聊天、協作、監控等場景。ASGI 是異步 Web 標準，配合 Uvicorn 服務器和 Django Channels，可實現 Django 的 WebSocket 功能，提升實時應用性能。

一、實時通信的核心：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()),
]

您正在閱讀的是《Django從入門到實戰》專欄！關注不迷路~

posted @ 2025-07-24 14:35 小王子1024 閱讀(421) 評論(0) 收藏舉報

刷新頁面返回頂部