### Django 4 中 WebSocket 的启动配置教程 要在 Django 4 中成功启动并使用 WebSocket，可以通过 `Django Channels` 来实现完整的 WebSocket 功能支持。以下是关于如何在 Django 4 中配置和启动 WebSocket 的详细介绍。 #### 1. 安装依赖项 为了使 WebSocket 正常工作，需要安装 `channels` 和 `channels_redis` 库。这些库提供了必要的功能来处理 WebSocket 连接以及利用 Redis 作为消息队列[^3]。 ```bash pip install channels channels_redis redis ``` #### 2. 修改项目的 settings.py 文件 在 `settings.py` 文件中添加以下配置： - **INSTALLED_APPS**: 将 `channels` 添加到已安装的应用列表中。 - **ASGI_APPLICATION**: 设置 ASGI 路由文件的位置。 - **CHANNEL_LAYERS**: 配置 Redis 作为通道层的后端。 ```python # settings.py INSTALLED_APPS = [ ... 'channels', ] ASGI_APPLICATION = "your_project_name.asgi.application" CHANNEL_LAYERS = { "default": { "BACKEND": "channels_redis.core.RedisChannelLayer", "CONFIG": { "hosts": [("127.0.0.1", 6379)], # Redis 地址 }, }, } ``` 上述代码片段定义了 `ASGI_APPLICATION` 变量指向 `asgi.py` 文件中的应用实例，并设置了 Redis 作为消息队列的存储位置[^5]。 #### 3. 创建 asgi.py 文件 确保项目根目录下存在 `asgi.py` 文件，并按照如下方式进行修改： ```python import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack import your_app_name.routing # 替换为实际应用名称 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings') application = ProtocolTypeRouter({ "http": get_asgi_application(), "websocket": AuthMiddlewareStack( URLRouter( your_app_name.routing.websocket_urlpatterns # 替换为实际路由路径 ) ), }) ``` 此部分代码实现了 HTTP 请求与 WebSocket 请求之间的多路复用，并指定了 WebSocket 的路由规则。 #### 4. 创建 routing.py 文件 在具体的应用程序目录下创建一个名为 `routing.py` 的文件，用于定义 WebSocket 的路由规则。 ```python # your_app_name/routing.py from django.urls import re_path from . import consumers websocket_urlpatterns = [ re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()), ] ``` 这里定义了一个简单的 WebSocket 路由模式，匹配 `/ws/chat/<room_name>/` 形式的 URL 并将其映射到指定的消费者类[^2]。 #### 5. 编写 Consumer 类 在应用程序目录下的 `consumers.py` 文件中编写具体的 WebSocket 消费者逻辑。 ```python # your_app_name/consumers.py import json from channels.generic.websocket import AsyncWebsocketConsumer class ChatConsumer(AsyncWebsocketConsumer): async def connect(self): self.room_name = self.scope['url_route']['kwargs']['room_name'] self.room_group_name = f'chat_{self.room_name}' await self.channel_layer.group_add( self.room_group_name, self.channel_name ) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard( self.room_group_name, self.channel_name ) async def receive(self, text_data=None, bytes_data=None): text_data_json = json.loads(text_data) message = text_data_json['message'] await self.channel_layer.group_send( self.room_group_name, { 'type': 'chat_message', 'message': message } ) async def chat_message(self, event): message = event['message'] await self.send(text_data=json.dumps({ 'message': message })) ``` 这段代码展示了如何通过异步方法接收、发送和广播消息给同一房间内的其他客户端。 #### 6. 启动开发服务器 完成以上配置后，在终端运行以下命令以启动 Django 开发服务器： ```bash python manage.py runserver ``` 如果一切配置无误，则可以在浏览器中访问 `http://localhost:8000/chat/` 页面测试 WebSocket 功能。 --- ###