Django 3.0+ 原生 WebSockets 实现指南：基于 ASGI 的全栈开发

Django 3.0 原生 WebSockets 实现指南

随着 Django 3.0 的发布，框架已原生支持 ASGI（异步服务器网关接口），这意味着在 Django 应用中集成 WebSockets 不再需要引入任何第三方依赖库。本文将深入讲解如何通过扩展默认的 ASGI 应用程序来利用 Django 处理 WebSocket 连接。内容涵盖在示例 ASGI 应用中处理连接、数据发送与接收，以及实现具体业务逻辑的完整步骤。

环境准备与项目初始化

环境准备与项目初始化

首先，请确保您的计算机已安装 Python 3.6 或更高版本。Django 3.0 及后续版本强制要求使用 Python 3.6+，这是为了支持 async 和 await 关键字。完成 Python 环境配置后，创建一个新的项目目录并进入该目录。接着，在 virtualenv 虚拟环境中安装 Django，并在当前项目目录下初始化一个新的 Django 应用。

$ mkdir django_websockets && cd django_websockets
$ python -m venv venv
$ source venv/bin/activate
$ pip install django
$ django-admin startproject websocket_app .

进入生成的 Django 应用目录（例如 websocket_app），您会看到一个名为 asgi.py 的文件。该文件包含了 Django 默认的 ASGI 配置，其基本内容如下所示：

import os
 
from django.core.asgi import get_asgi_application
 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'websocket_app.settings')
 
application = get_asgi_application()

此文件导出了名为 application 的默认 Django ASGI 应用程序，您可以使用 uvicorn 或 daphne 等 ASGI 服务器来运行它。在深入具体实现之前，有必要先了解 ASGI 应用程序的基本结构和工作原理。

ASGI 应用结构解析

ASGI 应用结构解析

ASGI（Asynchronous Server Gateway Interface，异步服务器网关接口）是构建 Python 异步 Web 服务的标准规范。作为 WSGI 的精神继承者，ASGI 已被 Django、Flask 等主流框架长期采用。ASGI 的核心优势在于它允许开发者利用 Python 原生的 async/await 语法来构建支持长连接的服务，例如 WebSockets 和服务端发送事件（Server-Sent Events）。

一个标准的 ASGI 应用程序是一个异步函数，接收三个参数：scope（当前请求的上下文字典）、receive（一个异步函数，用于监听传入事件）和 send（一个异步函数，用于向客户端发送事件）。

在 ASGI 应用程序内部，您可以根据 scope 字典中的值对请求进行路由。例如，通过检查 scope['type'] 的值，可以判断当前请求是 HTTP 请求还是 WebSocket 请求。要监听客户端数据，需等待 receive 函数返回；当需要向客户端发送数据时，调用 send 函数并将数据作为参数传入即可。接下来，我们将通过示例代码展示这一机制的具体应用。

构建自定义 ASGI 应用

构建自定义 ASGI 应用

在 asgi.py 文件中，我们将使用自定义的 ASGI 应用程序包装 Django 默认的 ASGI 功能，以便自主处理 WebSocket 连接。具体做法是定义一个名为 application 的异步函数，该函数必须包含三个 ASGI 参数：scope、receive 和 send。我们将 get_asgi_application() 的返回值重命名为 django_application，因为我们需要它专门处理 HTTP 请求。在 application 函数内部，我们将检查 scope['type'] 的值以确定请求类型：如果类型为 'http'，则交由 Django 处理；如果类型为 'websocket'，则由我们自定义的逻辑处理。最终的 asgi.py 文件结构如下：

import os
 
from django.core.asgi import get_asgi_application
 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'websocket_app.settings')
 
django_application = get_asgi_application()
 
async def application(scope, receive, send):
  if scope['type'] == 'http':
    # Let Django handle HTTP requests
    await django_application(scope, receive, send)
  elif scope['type'] == 'websocket':
    # We'll handle Websocket connections here
    pass
  else:
    raise NotImplementedError(f"Unknown scope type {scope['type']}")

接下来，我们需要创建一个专门处理 WebSocket 连接的函数。在与 asgi.py 相同的目录下创建 websocket.py 文件，并定义一个名为 websocket_application 的 ASGI 应用程序函数，同样接受 scope、receive 和 send 三个参数。随后，在 asgi.py 中导入 websocket_application，并在 application 函数内部调用它来处理 WebSocket 请求，传入相应的 scope、receive 和 send 参数。代码结构如下：

# asgi.py
import os
 
from django.core.asgi import get_asgi_application
from websocket_app.websocket import websocket_application
 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'websocket_app.settings')
 
django_application = get_asgi_application()
 
async def application(scope, receive, send):
  if scope['type'] == 'http':
    await django_application(scope, receive, send)
  elif scope['type'] == 'websocket':
    await websocket_application(scope, receive, send)
  else:
    raise NotImplementedError(f"Unknown scope type {scope['type']}")
 
# websocket.py
async def websocket_application(scope, receive, send):
  pass

接下来，让我们为 WebSocket 应用程序实现具体的业务逻辑。我们将监听所有 WebSocket 连接，当客户端发送字符串“ping”时，服务器将响应字符串“pong!”。

在 websocket_application 函数内部，我们将定义一个无限循环（while True），该循环将持续处理 WebSocket 请求，直到连接关闭。在循环内部，我们将等待服务器从客户端接收到的任何新事件。随后，根据事件的内容执行相应操作，并将响应发送回客户端。

首先处理连接建立阶段。当新的 WebSocket 客户端连接到服务器时，服务器会收到 'websocket.connect' 事件。为了允许该连接，我们需要发送一个 'websocket.accept' 事件作为响应。这将完成 WebSocket 握手过程，并与客户端建立持久连接。

同时，我们还需要处理断开连接事件。当客户端终止与服务器的连接时，服务器会收到 'websocket.disconnect' 事件。在此处，我们可以跳出无限循环，从而优雅地结束该连接的处理流程。

最后，处理来自客户端的数据请求。为此，我们需要监听 'websocket.receive' 事件。当收到 'websocket.receive' 事件时，检查 event['text'] 的值是否为“ping”。如果是，则发送一个 'websocket.send' 事件，其 text 字段值为“pong!”。

完成上述 WebSocket 逻辑后，我们的 websocket.py 文件完整代码应如下所示：

# websocket.py
async def websocket_application(scope, receive, send):
  while True:
    event = await receive()
 
    if event['type'] == 'websocket.connect':
      await send({
        'type': 'websocket.accept'
      })
    
    if event['type'] == 'websocket.disconnect':
      break
    
    if event['type'] == 'websocket.receive':
      if event['text'] == 'ping':
        await send({
          'type': 'websocket.send',
          'text': 'pong!'
        })

测试与验证

测试与验证

现在，我们的 ASGI 应用程序已配置为处理 WebSocket 连接，并且实现了基本的 WebSocket 服务器逻辑。接下来进行功能测试。需要注意的是，Django 默认的开发服务器（./manage.py runserver）目前不支持直接运行 asgi.py 文件，因此无法通过常规方式测试连接。您需要使用 ASGI 服务器（如 uvicorn）来运行该应用。首先安装 uvicorn：

$ pip install uvicorn

安装 uvicorn 后，可以使用以下命令启动 ASGI 应用程序：

$ uvicorn websocket_app.asgi:application
INFO:   Started server process [25557]
INFO:   Waiting for application startup.
INFO:   ASGI 'lifespan' protocol appears unsupported.
INFO:   Application startup complete.
INFO:   Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

为了测试 WebSocket 连接，请打开浏览器开发者工具的新标签页。在控制台中，创建一个名为 ws 的 WebSocket 实例，指向 ws://localhost:8000/。接着，为 ws 附加 onmessage 处理程序，用于将 event.data 记录到控制台。最后，调用 ws.send('ping') 向服务器发送消息。如果在控制台中看到“pong!”，则说明测试成功。

> ws = new WebSocket('ws://localhost:8000/')
 WebSocket {url: "ws://localhost:8000/", readyState: 0, bufferedAmount: 0, onopen: null, onerror: null, …}
> ws.onmessage = event => console.log(event.data)
 event => console.log(event.data)
> ws.send("ping")
 undefined
 pong!

恭喜！您现在已掌握如何使用 ASGI 将 WebSocket 支持添加到 Django 应用程序中。您可以利用这一技术构建各种实时交互应用。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。