Django3项目中使用websocket

2021年5月16日 22:24 阅读 271 评论 0

前言

最近项目中用到了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
微信
支付宝
登录后即可进行评论/回复
文章目录