前言
最近项目中用到了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
这样我们的服务就启动了,如下图:
测试
打开浏览器,F12打开开发者工具,在console依次输入以下内容进行测试:
ws = new WebSocket('ws://localhost:8000/')
ws.onmessage = event => console.log(event.data)
ws.send("ping")
如图所示,websocket连接创建成功且消息首发也是成功的。
恭喜,你现在知道了怎么在Django项目中使用websocket,可以用他写出很多有意思的内容!
Django版权声明:如无特殊说明,文章均为本站原创,转载请注明出处
本文链接:https://www.yangyingqi.com/48.html