一、概述

现在Django 3.0附带了对ASGI的支持，将Websockets添加到Django应用中不需要任何额外的依赖关系。 在本文中，您将学习如何通过扩展默认的ASGI应用程序来使用Django处理Websocket。 我们将介绍如何在示例ASGI应用程序中处理Websocket连接，发送和接收数据以及实现业务逻辑。

注意：Django 3.0不支持dwebsocket模块，启动时，会报错：

TypeError: WebSocketMiddleware() takes no arguments

因此，如果使用Django 3.0，必须使用channels

channels介绍

channels是以django插件的形式存在，它不仅能处理http请求，还提供对websocket、MQTT等长连接支持。不仅如此，channels在保留了原生django的同步和易用的特性上还带来了异步处理方式(channels2.X版本)，并且将django自带的认证系统以及session集成到模块中，扩展性非常强。官方文档：https://channels.readthedocs.io/en/latest/index.html

安装以及安装需求

channels2.0最低django版本要求是1.11+，python3.5+

本文采用的是python 3.7.3，django 3.1

pip3 install channels

二、开始使用

环境说明

Django==3.1
channels==2.4.0
paramiko==2.7.2
uvicorn==0.11.8

新建项目：django3_websocket，应用名称：web

看一下与settings.py同级目录。 您应该看到一个名为asgi.py的文件。 其内容如下所示：

import osfrom django.core.asgi import get_asgi_applicationos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django3_websocket.settings')application = get_asgi_application()

该文件提供了默认的Django ASGI设置，并公开了一个名为application的ASGI应用程序，可以使用uvicorn或daphne等ASGI服务器运行该应用程序。 在进一步介绍之前，让我们看一下ASGI应用程序的结构。

ASGI应用程序结构

ASGI或“异步服务器网关接口”是用于使用Python构建异步Web服务的规范。它是WSGI的精神继承者，WSGI已被Django和Flask等框架使用了很长时间。 ASGI使您可以使用Python的本机异步/等待功能来构建支持长期连接的Web服务，例如Websockets和Server Sent Events。

ASGI应用程序是一个异步函数，它带有3个参数：作用域（当前请求的上下文），接收（一个异步函数，可让您侦听传入的事件）和发送（一个异步函数，可将事件发送至客户端）。

在ASGI应用程序内部，您可以根据范围字典中的值路由请求。例如，您可以通过检查scope [‘type’]的值来检查该请求是HTTP请求还是Websocket请求。要侦听来自客户端的数据，您可以等待接收功能。准备好将数据发送到客户端时，可以等待发送功能，然后将要发送给客户端的任何数据传递给客户端。让我们看一下这在示例应用程序中是如何工作的。

创建一个ASGI应用

在我们的asgi.py文件中，我们将使用我们自己的ASGI应用程序包装Django的默认ASGI应用程序功能，以便自己处理Websocket连接。为此，我们需要定义一个名为application的异步函数，该函数需要3个ASGI参数：scope，receive和send。将get_asgi_application调用的结果重命名为django_application，因为我们需要它处理HTTP请求。在我们的应用程序函数内部，我们将检查scope [‘type’]的值以确定请求类型。如果请求类型为“ http”，则该请求为普通的HTTP请求，我们应该让Django处理它。如果请求类型为“ websocket”，那么我们将自己处理逻辑。

在views.py的同级目录，创建文件asgi.py，内容如下：

# ！/usr/bin/python3
# -*- coding: utf-8 -*-
import osfrom django.core.asgi import get_asgi_application# 注意修改项目名
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django3_websocket.settings')django_application = get_asgi_application()async def application(scope, receive, send):if scope['type'] == 'http':# Let Django handle HTTP requestsawait django_application(scope, receive, send)elif scope['type'] == 'websocket':# We'll handle Websocket connections herepasselse:raise NotImplementedError(f"Unknown scope type {scope['type']}")

在views.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':breakif event['type'] == 'websocket.receive':if event['text'] == 'ping':await send({'type': 'websocket.send','text': 'pong!'})

现在，我们需要创建一个函数来处理Websocket连接。 在与asgi.py文件相同的文件夹中创建一个名为websocket.py的文件，并定义一个名为websocket_application的ASGI应用程序函数，该函数接受3个ASGI参数。 接下来，我们将在我们的asgi.py文件中导入websocket_application，并在我们的应用程序函数内部调用它来处理Websocket请求，传入范围，接收和发送参数。 它看起来应该像这样：

import osfrom django.core.asgi import get_asgi_application
# 注意修改应用名
from web.websocket import websocket_applicationos.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应用程序实现一些逻辑。我们将监听所有Websocket连接，当客户端发送字符串“ ping”时，我们将以字符串“ pong！”进行响应。

在websocket_application函数内部，我们将定义一个不确定的循环，该循环将处理Websocket请求，直到关闭连接。在该循环内，我们将等待服务器从客户端收到的任何新事件。然后，我们将根据事件的内容采取行动，并将响应发送给客户端。

首先，让我们处理连接。当新的Websocket客户端连接到服务器时，我们将收到“ websocket.connect”事件。为了允许这种连接，我们将发送一个“ websocket.accept”事件作为响应。这将完成Websocket握手并与客户端建立持久连接。

当客户端终止其与服务器的连接时，我们还需要处理断开连接事件。为此，我们将监听“ websocket.disconnect”事件。当客户端断开连接时，我们将摆脱不确定的循环。

最后，我们需要处理来自客户端的请求。为此，我们将监听“ websocket.receive”事件。当我们从客户端收到“ websocket.receive”事件时，我们将检查event [‘text’]的值是否为“ ping”。如果是，我们将发送一个’websocket.send’事件，其文本值为’pong！’。

测试

现在，我们的ASGI应用程序已设置为处理Websocket连接，并且我们已经实现了Websocket服务器逻辑，让我们对其进行测试。 目前，Django开发服务器不使用asgi.py文件，因此您将无法使用./manage.py runserver测试连接。 相反，您需要使用ASGI服务器（例如uvicorn）运行该应用程序。 让我们安装它：

pip3 install uvicorn

安装uvicorn后，我们可以使用以下命令运行ASGI应用程序：

注意：打开cmd控制台，切换到项目django3_websocket目录，执行命令

uvicorn web.asgi:application

输出：

?[32mINFO?[0m:     Started server process [?[36m15320?[0m]
?[32mINFO?[0m:     Waiting for application startup.
?[32mINFO?[0m:     ASGI 'lifespan' protocol appears unsupported.
?[32mINFO?[0m:     Application startup complete.
?[32mINFO?[0m:     Uvicorn running on ?[1mhttp://127.0.0.1:8000?[0m (Press CTRL+C to quit)

如果出现以下报错：

    'DIRS': [os.path.join(BASE_DIR, 'templates')]
NameError: name 'os' is not defined

请修改settings.py，在顶头部分，导入os模块。

import os

再次执行uvicorn 命令即可。

访问页面

http://127.0.0.1:8000

效果如下：

要测试Websocket连接，请在新选项卡中打开浏览器的开发工具。 在控制台中，创建一个名为ws的新Websocket实例，该实例指向ws：// localhost：8000 /。 然后将onmessage处理程序附加到将event.data记录到控制台的ws。 最后，调用ws.send（’ping’）将消息发送到服务器。 您应该看到值“ pong！”。 登录到控制台。

请依次输入以下命令：

ws = new WebSocket('ws://localhost:8000/')
ws.onmessage = event => console.log(event.data)
ws.send("ping")

效果如下：

恭喜！ 现在，您知道了如何使用ASGI将Websocket支持添加到Django应用程序中。 去用它来制作很棒的东西。

本文参考链接：

https://www.mindg.cn/?p=2489