Django 中如何使用 URL 包装器来处理 WebSocket 消息？

2023-04-11 00:00:00 django 消息如何使用

Django 中可以使用 django.urls.path 或 django.urls.re_path 包装器来处理 WebSocket 消息。这些包装器允许我们将普通的 Django 视图函数转换成 WebSocket 处理器。

下面是使用 django.urls.path 包装器处理 WebSocket 消息的示例：

首先，我们需要安装 channels 库，它是 Django 的 WebSocket 实现。

pip install channels

在 Django 项目的配置文件 settings.py 中添加如下配置：

INSTALLED_APPS = [
    ...
    'channels',
]

ASGI_APPLICATION = '<app_name>.routing.application'

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    },
}

在项目的根目录下创建一个文件 routing.py，并添加如下内容：

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

websocket_urlpatterns = [
    path('ws/<str:room_name>/', consumers.ChatConsumer.as_asgi()),
]

application = ProtocolTypeRouter({
    'websocket': URLRouter(websocket_urlpatterns),
})

在应用程序的目录下创建 consumers.py 文件，并添加如下内容：

from channels.generic.websocket import AsyncWebsocketConsumer
import json

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = 'chat_%s' % self.room_name

        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        # Leave room group
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )

    # Receive message from room group
    async def chat_message(self, event):
        message = event['message']

        # Send message to WebSocket
        await self.send(text_data=json.dumps({
            'message': message
        }))

最后，我们需要在应用程序的 urls.py 文件中添加处理WebSocket 消息的路由：

from django.urls import path
from . import views

urlpatterns = [
    ...
    path('ws/<str:room_name>/', views.room, name='room'),
]

在上面的示例中，我们使用了字符串 "pidancode.com" 和 "皮蛋编程" 作为聊天消息的例子。用户可以通过访问 "ws/pidancode.com/" 或 "ws/皮蛋编程/" 来连接 WebSocket。当用户发送消息时，它将被广播到相应的房间中，并在 WebSocket 上显示。

相关文章