Django3项目中使用websocket

前言

最近项目中用到了websocket，看了文档，准备使用dwebsocket，但是报了错。

TypeError: WebSocketMiddleware() takes no arguments

网上查阅一番之后，原来是Django 3.0不支持dwebsocket模块，如果使用Django 3.0，必须使用channels。

经过一番折腾总算整出一个demo

安装channels

pip install channels

创建项目

这里推荐使用Pycharm创建项目，还需要new一个project选择Django即可，这里假设创建的项目名为my_project,其中新建一个app，名为test。

创建好项目后，我们可以看见项目同名文件夹my_project下有个asgi.py文件，里面内容如下：

import os 

from django.core.asgi import get_asgi_application 

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django3_websocket.settings') 

application = get_asgi_application()

该文件提供了默认的Django ASGI设置，并公开了一个名为application的ASGI应用程序，可以使用uvicorn或daphne等ASGI服务器运行该应用程序

创建ASGI应用

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

配置ASGI

首先我们修改asgi.py，内容如下：

# ！/usr/bin/python3 
# -*- coding: utf-8 -*- 
import os 
from django.core.asgi import get_asgi_application 

# 注意修改项目名 
import os 

from django.core.asgi import get_asgi_application 

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_project.settings') 

application = get_asgi_application() 
from apps.test.websocket import websocket_application 


async def application(scope, receive, send): 
    if scope['type'] == 'http': 
        # Let Django handle HTTP requests 
        await application(scope, receive, send) 
    elif scope['type'] == 'websocket': 
        # We'll handle Websocket connections here 
        await websocket_application(scope, receive, send) 
    else: 
        raise NotImplementedError(f"Unknown scope type {scope['type']}")

在名为test的app下创建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': # 收到数据时 
            text = event['text'] # 收到的数据 
            await send({ 
                'type': 'websocket.send', 
                'text': 'get' 
            })

启动ASGI

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

pip install uvicorn

安装成功后在terminal运行如下命令:

uvicorn web.asgi:application

这样我们的服务就启动了，如下图：启动uvicorn

测试

打开浏览器，F12打开开发者工具，在console依次输入以下内容进行测试：

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

测试websocket

如图所示，websocket连接创建成功且消息首发也是成功的。

恭喜，你现在知道了怎么在Django项目中使用websocket，可以用他写出很多有意思的内容！

最后修改于2021年5月16日 22:24

版权声明：如无特殊说明，文章均为本站原创，转载请注明出处

本文链接：https://www.yangyingqi.com/48.html

Django