如何在DRF+React前后端分离项目中实现WebSockets？

核心结论 #

依然需要借助Django Channels（结合python-socketio）来实现，因为DRF本身是REST框架，不支持WebSocket/SocketIO这类实时通信协议。Channels是Django官方推荐的实时扩展，能无缝对接DRF的ORM、认证、序列化器等生态，完美适配你前后端分离+事件驱动的需求。

具体实现步骤 #

1. 安装依赖包

pip install channels channels-redis python-socketio django-cors-headers

• channels：Django实时通信核心扩展
• channels-redis：用于分布式部署的通道层（本地开发可先用InMemoryChannelLayer，生产必须换Redis）
• python-socketio：提供SocketIO协议支持
• django-cors-headers：处理前后端跨域问题

2. 配置Django项目

修改settings.py：

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

# 配置CORS
CORS_ALLOWED_ORIGINS = [
    "http://localhost:3000",  # 你的React前端地址
]
CORS_ALLOW_CREDENTIALS = True

# 配置Channels通道层
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

# 指定ASGI应用入口
ASGI_APPLICATION = "your_project_name.asgi.application"

3. 编写SocketIO事件处理逻辑

创建一个处理事件的模块（比如apps/tasks/socket_handlers.py），对应前端的getTasks、addTask事件：

import socketio
from .models import Task
from .serializers import TaskSerializer
from rest_framework.authtoken.models import Token

# 初始化SocketIO服务，允许前端跨域连接
sio = socketio.AsyncServer(
    async_mode='asgi',
    cors_allowed_origins=["http://localhost:3000"]  # 生产环境替换为真实前端域名
)

# 处理前端的getTasks事件
@sio.event
async def getTasks(sid, data):
    # 验证用户身份（通过前端传递的token）
    try:
        token = data.get('token')
        user = Token.objects.get(key=token).user
    except Token.DoesNotExist:
        await sio.emit('authError', 'Invalid authentication token', room=sid)
        return

    # 从数据库获取用户的任务，用DRF序列化器处理数据
    tasks = Task.objects.filter(user=user)
    serializer = TaskSerializer(tasks, many=True)
    # 将结果发送给当前请求的客户端
    await sio.emit('tasksList', serializer.data, room=sid)

# 处理前端的addTask事件
@sio.event
async def addTask(sid, data):
    try:
        token = data.get('token')
        user = Token.objects.get(key=token).user
    except Token.DoesNotExist:
        await sio.emit('authError', 'Invalid authentication token', room=sid)
        return

    # 用DRF序列化器验证并保存新任务
    serializer = TaskSerializer(data=data['task'], context={'user': user})
    if serializer.is_valid():
        new_task = serializer.save()
        # 广播新任务给所有在线客户端（实时同步用）
        await sio.emit('taskAdded', serializer.data)
    else:
        # 发送错误信息给请求客户端
        await sio.emit('addTaskError', serializer.errors, room=sid)

4. 整合DRF与SocketIO服务

修改项目根目录的asgi.py，让DRF的HTTP请求和SocketIO请求共用同一个ASGI服务器：

import os
from django.core.asgi import get_asgi_application
import socketio
from apps.tasks.socket_handlers import sio

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

# 获取DRF的ASGI应用（处理HTTP请求）
django_asgi_app = get_asgi_application()

# 创建SocketIO的ASGI应用，将DRF应用作为 fallback
socketio_app = socketio.ASGIApp(sio, django_asgi_app)

# 最终的ASGI入口：优先处理SocketIO请求，其余交给DRF
application = socketio_app

5. 启动ASGI服务器

用uvicorn运行服务器（Django默认的runserver不支持ASGI）：

uvicorn your_project_name.asgi:application --reload --port 8000

React前端对接示例 #

安装SocketIO客户端：

npm install socket.io-client

在React组件中使用：

import { useEffect, useState } from 'react';
import io from 'socket.io-client';

// 连接后端SocketIO服务
const socket = io('http://localhost:8000');

function TaskManager() {
  const [tasks, setTasks] = useState([]);
  const [newTaskTitle, setNewTaskTitle] = useState('');
  const authToken = localStorage.getItem('authToken');

  useEffect(() => {
    // 监听后端返回的tasksList事件，更新任务列表
    socket.on('tasksList', (taskData) => {
      setTasks(taskData);
    });

    // 监听新任务添加成功的广播，更新本地列表
    socket.on('taskAdded', (newTask) => {
      setTasks(prev => [...prev, newTask]);
    });

    // 监听错误事件
    socket.on('authError', (msg) => alert(msg));
    socket.on('addTaskError', (errors) => console.error(errors));

    // 页面加载时请求任务列表
    socket.emit('getTasks', { token: authToken });

    // 组件卸载时断开连接
    return () => socket.disconnect();
  }, [authToken]);

  const handleAddTask = () => {
    if (!newTaskTitle) return;
    // 触发后端的addTask事件
    socket.emit('addTask', {
      token: authToken,
      task: { title: newTaskTitle }
    });
    setNewTaskTitle('');
  };

  return (
    <div>
      <input
        type="text"
        value={newTaskTitle}
        onChange={(e) => setNewTaskTitle(e.target.value)}
        placeholder="输入新任务"
      />
      <button onClick={handleAddTask}>添加任务</button>
      <ul>
        {tasks.map(task => (
          <li key={task.id}>{task.title}</li>
        ))}
      </ul>
    </div>
  );
}

export default TaskManager;

关键注意事项 #

• 跨域配置：务必在后端python-socketio和django-cors-headers中同时配置允许的前端域名，避免跨域错误。
• 用户认证：通过SocketIO的事件数据传递DRF的认证token，后端验证后再处理业务逻辑，确保数据安全。
• 生产部署：必须使用Redis作为Channels的通道层，不能用内存层；同时ASGI服务器推荐用gunicorn配合uvicorn worker。

内容的提问来源于stack exchange，提问作者Ch1irxo 17