Django+DRF中如何实现POST接口与WebSocket接口的消息互通

Django默认的同步请求响应模型不支持WebSocket长连接，要实现POST接口收到请求后触发WebSocket广播，标准方案是搭配Django Channels扩展，通过Channel Layer完成HTTP接口和WebSocket通道之间的消息中转，具体步骤如下：

1. 依赖安装与基础配置 #

• 安装必需依赖

pip install channels channels_redis djangorestframework

• 修改项目settings.py配置
  注意channels必须放在注册应用列表的最前面，否则会出现路由覆盖问题：

INSTALLED_APPS = [
    "channels",
    "rest_framework",
    # 你的其他业务app
]

# 指定ASGI启动入口，替换默认WSGI配置
ASGI_APPLICATION = "你的项目根目录名.asgi.application"

# 配置Channel Layer，生产环境必须用Redis做后端，本地调试可以临时用内存层
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

• 修改项目根目录下的asgi.py，分别挂载HTTP和WebSocket路由：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from 你的业务app名.routing import websocket_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目根目录名.settings')

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

2. 编写WebSocket对应逻辑 #

以下代码匹配问题中提到的/wsEvens路径，如果是路径笔误可自行调整：

• 在业务app下新建consumers.py，编写WebSocket连接处理类：

import json
from channels.generic.websocket import AsyncWebsocketConsumer

class EventConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # 所有连接用户加入同一个公共分组，方便后续统一广播
        self.group_name = "public_event_group"
        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        # 连接断开时将当前连接移出分组
        await self.channel_layer.group_discard(
            self.group_name,
            self.channel_name
        )

    # 自定义消息处理方法，方法名必须和后续group_send传入的type字段完全匹配
    async def push_new_event(self, event):
        message_content = event["data"]
        sender_id = event["sender"]
        # 跳过消息发送者本身，只推送给其他在线用户
        if self.scope["user"].id != sender_id:
            await self.send(text_data=json.dumps({
                "event_type": "new_event",
                "payload": message_content
            }))

• 在业务app下新建routing.py，配置WebSocket路由：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r"wsEvens/$", consumers.EventConsumer.as_asgi()),
]

3. 编写POST /createEvent DRF接口 #

核心逻辑是在接口完成事件数据校验、存储后，调用Channel Layer向公共分组发送广播消息，所有组内的WebSocket连接都会自动收到消息。

• 编写DRF视图：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from .serializers import EventSerializer # 自行编写对应事件模型的序列化器

class CreateEventView(APIView):
    def post(self, request):
        serializer = EventSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        saved_instance = serializer.save(creator=request.user)
        # 获取全局Channel Layer实例
        channel_layer = get_channel_layer()
        # 向公共分组发送广播
        async_to_sync(channel_layer.group_send)(
            "public_event_group", # 和消费者中定义的分组名保持一致
            {
                "type": "push_new_event", # 和消费者中定义的接收方法名保持一致
                "data": EventSerializer(saved_instance).data,
                "sender": request.user.id
            }
        )
        return Response(serializer.data, status=status.HTTP_201_CREATED)

• 将接口挂载到项目HTTP路由中：

from django.urls import path
from .views import CreateEventView

urlpatterns = [
    path("createEvent", CreateEventView.as_view(), name="create-event"),
    # 其他业务HTTP路由
]

踩坑提示 #

• 生产环境不要使用channels.layers.InMemoryChannelLayer内存层，多进程/多实例部署时内存层无法跨进程传递消息，会出现消息丢失
• group_send传入的type字段值必须和消费者类中定义的接收方法名完全对应，Channels会自动将下划线命名映射到类方法
• 部署时必须使用ASGI服务器（Daphne、Uvicorn等）启动项目，传统WSGI服务器（如默认配置的Gunicorn）无法处理WebSocket连接
• 本地调试前确保Redis服务正常启动，否则Channel Layer会抛出连接错误

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