使用 WebSockets 和 Django 通道在 Django 中实现实时功能

近年来，实时网络应用程序越来越受欢迎，用户无需手动刷新页面即可体验即时更新和互动功能。Django 作为一个强大的网络框架，提供了在应用程序中构建实时功能的必要工具。在本文中，我们将探讨如何利用 Django 通道和 WebSockets 创建动态和响应式网络应用程序。我们将深入探讨将实时功能无缝集成到 Django 项目中的实用示例和最佳实践。

Django Channels 简介

Django Channels 是一个项目，它扩展了 Django 处理异步通信的功能，包括 WebSockets、HTTP2 推送和后台任务。通过它，您可以在客户端和服务器之间实现双向通信，从而构建实时网络应用程序。

Django Channels 的核心是引入 “通道 “的概念，通道本质上是允许在应用程序不同部分之间发送消息的通信通道。通道可用于各种目的，如处理 WebSocket 连接、发送通知或执行后台任务。

设置 Django Channels

要开始使用 Django Channels，您需要在 Django 项目中安装它。您可以使用 pip 进行安装：

pip install channels

接下来，您需要配置 Django 项目以使用 Channels。更新 settings.py 文件，加入以下内容：

INSTALLED_APPS = [
    ...
    'channels',
]

ASGI_APPLICATION = 'myproject.asgi.application'

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('localhost', 6379)],
        },
    },
}

在本例中，我们假定 Redis 已在本地默认端口 (6379) 上运行。Django Channels 使用 Redis 作为通道层后端，以实现应用程序不同部分之间的通信。

创建 WebSocket 消费者

Django Channels 中的消费者类似于传统 Django 中的视图。它负责处理传入的 WebSocket 连接和消息。让我们创建一个简单的 WebSocket 消费者，它可以回传接收到的消息：

from channels.generic.websocket import WebsocketConsumer

class EchoConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()

    def disconnect(self, close_code):
        pass

    def receive(self, text_data):
        self.send(text_data=text_data)

在本例中，EchoConsumer 类继承自 WebsocketConsumer。它定义了三个方法：

• connect()：当建立 WebSocket 连接时调用。我们只需使用 self.accept() 接受连接即可。
• disconnect()：当建立 WebSocket 连接时调用： 当关闭 WebSocket 连接时调用。在这种情况下，我们不需要执行任何特定操作。
• receive()：从客户端接收到信息时调用。我们使用 self.send() 回传收到的 text_data。

路由 WebSocket 连接

要将传入的 WebSocket 连接路由到相应的消费者，需要定义路由配置。在项目中创建一个新文件 routing.py：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/echo/$', consumers.EchoConsumer.as_asgi()),
]

在本例中，我们使用 re_path() 定义了一个 WebSocket URL 模式。ws/echo/ 模式映射到我们之前创建的 EchoConsumer。as_asgi() 方法用于将消费者类转换为 ASGI 应用程序。

更新项目的 asgi.py 文件以包含 WebSocket 路由：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from . import routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

application = ProtocolTypeRouter({
    'http': get_asgi_application(),
    'websocket': URLRouter(routing.websocket_urlpatterns),
})

此配置可设置 ASGI 应用程序同时处理 HTTP 和 WebSocket 协议。ProtocolTypeRouter 根据协议类型路由传入的请求，而 URLRouter 则将 WebSocket URL 映射到各自的用户。

处理 WebSocket 消息

现在我们已经完成了基本设置，让我们来探讨一下如何在消费者中处理 WebSocket 消息。让我们创建一个聊天应用程序，客户可以向聊天室发送消息，所有连接的客户都能实时接收消息。

首先，更新 EchoConsumer 以处理聊天信息：

from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_sync

class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'
        
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name,
            self.channel_name
        )
        
        self.accept()

    def disconnect(self, close_code):
        async_to_sync(self.channel_layer.group_discard)(
            self.room_group_name,
            self.channel_name
        )

    def receive(self, text_data):
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': text_data
            }
        )

    def chat_message(self, event):
        message = event['message']
        self.send(text_data=message)

在此更新的消费者：

• connect()：从 WebSocket URL 中提取 room_name，并构建一个唯一的 room_group_name。我们使用 async_to_sync 同步调用 group_add()方法，将当前通道添加到房间组。
• disconnect()：当 WebSocket 连接关闭时，使用 group_discard()将频道从房间组中删除。
• receive()：从客户端接收到信息后，我们会使用 group_send() 将其发送到整个房间组。信息会以自定义事件类型 “chat_message “发送。
• chat_message()：这是一个自定义事件处理程序，在收到 “chat_message “事件时被调用。它会将收到的信息发送回客户端。

更新 routing.py 文件以映射新的 WebSocket URL 模式：

websocket_urlpatterns = [ 
    re_path( r'ws/chat/(?P\w+)/$' ,consumers.ChatConsumer.as_asgi()), 
]

URL 模式中的 room_name 参数允许客户端连接到不同的聊天室。

客户端 WebSocket 通信

要从客户端与 WebSocket 服务器交互，可以使用 JavaScript。下面是一个如何建立 WebSocket 连接并发送/接收信息的示例：

<script>
    const roomName = 'lobby';
    const chatSocket = new WebSocket(`ws://${window.location.host}/ws/chat/${roomName}/`);

    chatSocket.onmessage = function(event) {
        const message = event.data;
        console.log('Received message:', message);
        // Display the message in the chat UI
    };

    function sendMessage(message) {
        chatSocket.send(message);
    }

</script>

在本例中

• 创建了一个新的 WebSocket 实例，指定了带有所需房间名称的 WebSocket URL。
• 每当收到来自服务器的消息时，就会调用 onmessage 事件处理程序。您可以使用该处理程序在聊天用户界面中显示收到的消息。
• sendMessage() 函数通过 WebSocket 连接向服务器发送消息。

最佳实践和注意事项

使用 Django 通道和 WebSockets 构建实时应用程序时，请考虑以下最佳实践和注意事项：

• 验证和授权：确保 WebSocket 连接经过正确的身份验证和授权。您可以使用 Django 的身份验证系统来确保 WebSocket 端点的安全，并限制只有授权用户才能访问。

• 扩展和性能：随着应用程序的增长，您可能需要扩展 WebSocket 基础设施以处理增加的流量。请考虑部署多个通道层，并在它们之间平衡 WebSocket 连接的负载。此外，还要优化 WebSocket 处理程序，尽量减少阻塞操作，确保高效的消息处理。

• 错误处理和重新连接：优雅地处理 WebSocket 错误和断开连接。实施适当的错误处理机制，并在出现临时网络问题时为客户端提供自动重新连接的方法。

• 测试和调试：为您的 WebSocket 消费者和实时功能编写全面的测试。Django Channels 提供测试工具来模拟 WebSocket 连接并验证消费者的行为。使用 Django 调试工具栏等调试工具监控和调试实时应用程序。

• 安全考虑因素：保护您的 WebSocket 端点免受潜在安全漏洞的影响。对传入信息进行验证和消毒以防止 XSS 攻击，并实施速率限制以减轻潜在的 DoS 攻击。定期更新您的依赖项，包括 Django 通道，以确保您拥有最新的安全补丁。

结论

Django 通道和 WebSockets 是构建实时网络应用程序的强大组合。通过利用 Django 的异步功能和 WebSockets 提供的双向通信，您可以为用户创建动态的交互式体验。

在本文中，我们探讨了 Django 通道的基本原理，包括设置必要的依赖关系、创建 WebSocket 消费者和处理实时消息。我们演示了构建简单的 echo server 和聊天应用程序的实际示例。

作者：Simeon Emanuilov
原文：https://unfoldai.com/real-time-functionality-in-django-with-websockets-and-django-channels/