django3 websockets


一、概述

現在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 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應用程序結構

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 os

from 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 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']}")

 

在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':
            break

        if 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 os

from django.core.asgi import get_asgi_application
# 注意修改應用名
from web.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應用程序實現一些邏輯。我們將監聽所有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

 


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM