Python伺服器程式設計：使用django-channels實作WebSockets

在現今的Web開發中，即時通訊是不可或缺的功能之一。由於HTTP協議是一種請求-回應協議，因此使用HTTP的傳統方式來實現即時通訊非常不方便。 WebSockets協定是一種新興的協議，它為網路應用程式提供了即時雙向通訊功能，而且可以在同一個連接上發送和接收數據，非常適合即時應用程式。在Python伺服器程式設計中，可以使用django-channels框架來輕鬆實作WebSockets。

1. 安裝django-channels

在開始使用django-channels之前，首先需要安裝它。可以使用pip來安裝：

pip install channels

2. 建立Django專案

接下來，建立一個Django專案。在Django 2.x以上版本中，可以使用命令列工具建立專案：

django-admin startproject myproject

3. #設定django-channels

在安裝django-channels之後，需要將其添加到Django項目中。開啟settings.py文件，並在INSTALLED_APPS中加入'channels'。此外，還需要為django-channels配置一些設定：

# settings.py

# 添加channels到INSTALLED_APPS
INSTALLED_APPS = [
    # ...
    'channels',
]

# 配置channels
ASGI_APPLICATION = 'myproject.routing.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    },
}

上面的程式碼中，ASGI_APPLICATION指定了ASGI應用程式的入口點，而CHANNEL_LAYERS指定了預設頻道層的類型和參數。在這個例子中，使用了InMemoryChannelLayer，這是一個實作了簡單記憶體儲存的通道層。

4. 建立路由

在建立django-channels應用程式之前，需要先建立一個路由來路由傳入的WebSocket請求。路由是一個映射表，它將URL路徑映射到特定的Consumer類別。在Django中，路由通常定義在urls.py檔案中，但在django-channels中，由於它使用ASGI協議，因此將路由定義在routing.py中，如下所示：

# myproject/routing.py

from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path

application = ProtocolTypeRouter({
    # WebSocket使用的协议类型是“websocket”，将它放在第一位
    "websocket": URLRouter([
        path("ws/", MyConsumer.as_asgi()),
    ]),
})

上述程式碼中，我們使用ProtocolTypeRouter建立了一個協定路由，並設定了一個基於WebSocket的子路由。在這個範例中，WebSocket請求的URL是/ws/，並且連接時將使用MyConsumer類別。

5. 建立Consumer

在django-channels中，Consumer是一個處理網路請求的類別。可以在路由中將請求路由到consumer，然後consumer會處理請求並回傳回應。 Consumer一般由一個async def方法（即協程）實作。在建構Consumer時，必須繼承channels.generic.websocket.WebSocketConsumer類，並實作兩個方法：

• connect(self): 當WebSocket連線建立時，django-channels呼叫該方法。
• receive(self, text_data=None, bytes_data=None): 當從WebSocket連線接收到資料時，django-channels呼叫此方法。

以下是一個簡單的Consumer範例：

# myapp/consumers.py

import asyncio
import json

from channels.generic.websocket import AsyncWebsocketConsumer


class MyConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        """
        WebSocket连接建立时执行。
        """
        await self.accept()

    async def disconnect(self, code):
        """
        WebSocket连接中断时执行。
        """
        pass

    async def receive(self, text_data=None, bytes_data=None):
        """
        当从WebSocket连接接收到数据时执行。
        """
        # 解析WebSocket发送的JSON数据
        data = json.loads(text_data)
        # 从JSON数据中获取请求
        request = data['request']

        # 这里是处理请求的代码
        # ...

        # 发送响应到WebSocket连接
        response = {'status': 'OK', 'data': request}
        await self.send(json.dumps(response))

6. #啟動Django伺服器

現在，所有的設定都已完成，可以啟動Django伺服器並測試WebSocket連線了。在終端機中輸入以下指令來啟動Django伺服器：

python manage.py runserver

如果一切正常，應該可以透過http://127.0.0.1:8000/ws/來測試WebSocket連線了，如果連線成功，WebSocket Consumer的connect方法將會被執行。

總結：

使用django-channels實作WebSocket非常簡單，基本上只需要幾個步驟。需要注意的一點是，在django-channels應用程式中，經常使用asyncio協程，因此，需要使用Python 3.5及更高版本。此外，通道層的配置也很重要，如果想要使用持久化存儲，可以使用其他通道層，例如RedisChannelLayer等。

以上是Python伺服器程式設計：使用django-channels實作WebSockets的詳細內容。更多資訊請關注PHP中文網其他相關文章！