# dj_websocket

**Repository Path**: duans/dj_websocket

## Basic Information

- **Project Name**: dj_websocket
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-02
- **Last Updated**: 2026-03-15

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# django项目中实现websocket

在 Django 中实现 WebSocket，目前最主流、官方推荐的方式是使用 **Django Channels**。

Django 本身是基于 WSGI（同步）设计的，而 WebSocket 需要异步长连接支持。Django Channels 扩展了 Django，使其能够处理 WebSocket、HTTP2、后台任务等异步协议。

以下是基于 **Django 4.x/5.x** 和 **Channels 4.x** 的实现步骤：

## 1. 安装依赖

你需要安装 `channels` 和 `daphne`（ASGI 服务器）：

```shell
1pip install channels daphne
```

## 2. 修改`settings.py`

将 `channels` 添加到 `INSTALLED_APPS`，并配置 ASGI 应用入口。

```python
# settings.py
INSTALLED_APPS = [
    # ... 其他应用
    'channels',
]

# 指定 ASGI 应用入口 (默认是 dj_websocket.asgi:application)
ASGI_APPLICATION = 'dj_websocket.asgi.application'

# 配置通道层 (Channel Layer)
# 开发环境可以使用本地内存，生产环境强烈建议使用 Redis
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer"
        # 生产环境配置示例 (需安装 channels_redis):
        # "BACKEND": "channels_redis.core.RedisChannelLayer",
        # "CONFIG": {
        #     "hosts": [("127.0.0.1", 6379)],
        # },
    },
}
```

## 3. 创建路由 `routing.py`

在你的应用目录（例如 `chat/routing.py`）或项目根目录下定义 WebSocket 的路由映射。

```python
# chat/routing.py
from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    # 匹配 ws://host/ws/chat/<room_name>/
    re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()),
]
```

## 4. 编写消费者 `consumers.py`

消费者类似于 Django 的 View，但用于处理 WebSocket 事件（连接、断开、接收消息）。

```python
# chat/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async

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
        )

    # 接收来自 WebSocket 的消息
    async def receive(self, text_data):
        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
            }
        )

    # 接收来自群组的消息 (由 type="chat_message" 触发)
    async def chat_message(self, event):
        message = event["message"]

        # 发送消息给 WebSocket
        await self.send(text_data=json.dumps({
            "message": message
        }))
```

## 5. 配置项目级 `asgi.py`

在项目根目录（与 `settings.py` 同级）修改 `asgi.py`，将 WebSocket 路由整合进去。

```python
# myproject/asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from chat.routing import websocket_urlpatterns  # 导入刚才定义的路由

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

django_asgi_app = get_asgi_application()

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AuthMiddlewareStack(
        URLRouter(
            websocket_urlpatterns
        )
    ),
})
```

## 6. 前端连接测试

创建一个简单的 HTML 页面来测试连接：

```html
<script>
    const roomName = "test_room";
    // 注意协议是 ws:// 或 wss://
    const chatSocket = new WebSocket(
        'ws://' + window.location.host + '/ws/chat/' + roomName + '/'
    );

    chatSocket.onopen = function(e) {
        console.log("连接成功");
        // 发送消息
        chatSocket.send(JSON.stringify({'message': 'Hello Django!'}));
    };

    chatSocket.onmessage = function(e) {
        const data = JSON.parse(e.data);
        console.log("收到消息:", data.message);
        document.querySelector('#chat-log').value += (data.message + '\n');
    };

    chatSocket.onclose = function(e) {
        console.error("连接关闭");
    };
</script>
<textarea id="chat-log" cols="100" rows="20"></textarea>
```

## 7. 运行服务器

**重要**：运行 WebSocket 时，不能使用 `python manage.py runserver` (除非你只做了极简配置且 Django 版本较新自动检测，但通常建议显式指定)，也不能使用 Gunicorn (除非配合 gunicorn-websocket 插件，但不推荐)。

推荐使用 **Daphne** (Channels 自带的 ASGI 服务器) 或 **Uvicorn**。

### 开发环境运行：

```shell
# 使用daphne启动项目
daphne -b 0.0.0.0 -p 8000 dj_websocket.asgi:application
# 也可以使用uvicorn启动项目
uvicorn -b 0.0.0.0 -p 8000 dj_websocket.asgi:application
```

### 生产环境部署：

生产环境通常使用 **Daphne** 或 **Uvicorn** 配合 **Nginx** 反向代理。如果使用 Gunicorn，需要开启 `--worker-class uvicorn.workers.UvicornWorker`。

## 关键点总结

1. **异步支持**：Consumer 类通常继承自 `AsyncWebsocketConsumer`，使用 `async/await` 语法。
2. **Channel Layer**：如果你想实现群聊（广播消息给多人），必须配置 Channel Layer（开发用 InMemory，生产用 Redis）。如果是单对单且不需要跨进程通信，可以暂时不配 Redis，但为了架构扩展性，建议直接上 Redis。
3. **ASGI**：整个项目必须以 ASGI 模式运行，而不是传统的 WSGI。
4. **数据库操作**：在异步 Consumer 中直接调用 Django ORM 会报错。如果需要查库，需使用 `database_sync_to_async` 装饰器或将 ORM 调用包裹在 `sync_to_async` 中。