Django WebSockets 入门指南（Channels 实战）
用 Django Channels 实现 WebSocket：项目配置、编写 Consumer、消息广播与前端对接，一步步做出实时通知与聊天功能。
直接回答：普通短 HTTP 响应不提供持续双向通道；SSE、流式响应和 WebSocket 适用场景不同，Django Channels 补齐了这块拼图：它为 Django 加入 WebSocket（以及长轮询、后台任务）支持，让实时通知、协同编辑、即时聊天在 Django 里落地。
为什么需要 Channels
HTTP 的天然单向性意味着服务器不能主动找浏览器。WebSocket 建立全双工长连接后，服务端可以随时推送——消息提醒、进度更新、实时协作都建立在这之上。Channels 把这套能力以 Django 熟悉的风格（路由、消费者、中间件）呈现出来。
项目配置
pip install channels daphne channels-redis
# settings.py
INSTALLED_APPS = ["daphne", "channels", ...]
ASGI_APPLICATION = "myproject.asgi.application"
# 开发期可用内存层；生产用 Redis
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {"hosts": [("127.0.0.1", 6379)]},
}
}
Daphne 替代 WSGI 服务器接管 ASGI 流量——HTTP 和 WebSocket 都走它。
第一个 Consumer
Consumer 之于 WebSocket，相当于 View 之于 HTTP：
# consumers.py
import json
from channels.generic.websocket import WebsocketConsumer
class EchoConsumer(WebsocketConsumer):
def connect(self):
self.accept()
def receive(self, text_data):
data = json.loads(text_data)
self.send(text_data=json.dumps({"echo": data["msg"]}))
def disconnect(self, code):
pass
# routing.py
from django.urls import path
from .consumers import EchoConsumer
websocket_urlpatterns = [path("ws/echo/", EchoConsumer.as_asgi())]
还需连接 ASGI 路由（设置 DJANGO_SETTINGS_MODULE 后，先调用 get_asgi_application，再导入可能引用模型的 routing）：
# myproject/asgi.py
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
from django.core.asgi import get_asgi_application
django_asgi_app = get_asgi_application()
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator
from chat.routing import websocket_urlpatterns # chat 替换为你的应用包名
application = ProtocolTypeRouter({
"http": django_asgi_app,
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(URLRouter(websocket_urlpatterns))
),
})
前端对接：
const ws = new WebSocket("ws://localhost:8000/ws/echo/");
ws.onmessage = (e) => console.log(JSON.parse(e.data));
ws.onopen = () => ws.send(JSON.stringify({ msg: "你好" }));
广播给多个客户端
群聊/通知场景需要 Channel Layer 的分组机制：
from asgiref.sync import async_to_sync
class ChatConsumer(WebsocketConsumer):
def connect(self):
self.room = self.scope["url_route"]["kwargs"]["room"]
async_to_sync(self.channel_layer.group_add)(self.room, self.channel_name)
self.accept()
def receive(self, text_data):
async_to_sync(self.channel_layer.group_send)(
self.room,
{"type": "chat.message", "msg": json.loads(text_data)["msg"]},
)
def disconnect(self, code):
async_to_sync(self.channel_layer.group_discard)(self.room, self.channel_name)
def chat_message(self, event):
self.send(text_data=json.dumps({"msg": event["msg"]}))
聊天室片段还需配置含 room 参数的路由；入组前验证登录身份、房间权限、合法组名和消息长度。group_send 把消息扇出到组内连接——这就是聊天室与实时通知的核心原语。
生产部署要点
- ASGI 服务器（Daphne/Uvicorn）独立进程，与 WSGI 的 HTTP 流量分开或统一由 ASGI 接管；
- Channel Layer 生产用 Redis，内存层仅限开发；
- 长连接对负载均衡有要求：使用支持 WebSocket Upgrade 的代理并设置连接超时；纯 WebSocket 不必然要求会话保持；
- 连接数、消息速率是核心容量指标，上线前先压测；在连接建立、断开及消息处理处记录计数和耗时，分别检查连接容量与消息积压。
常见问题（FAQ）
Q：Channels 和单独的 WebSocket 服务怎么选？
A：少量实时功能、主站是 Django：Channels 最顺。大规模连接应先按消息负载压测，再评估专用网关，Django 只出业务 API。
Q：同步 Consumer 和 AsyncConsumer 选哪个？
A：IO 密集且并发高用 AsyncConsumer；要调 Django ORM 的同步代码用 WebsocketConsumer（自动线程池）或 database_sync_to_async 包装。
Q：连接鉴权怎么做？
A：Channels 中间件栈支持 Django session 认证；JWT 场景需自行校验令牌及房间权限；浏览器原生 WebSocket 不能任意设置请求头，避免把长期 token 放 URL。使用短期票据或安全会话，并校验 Origin。
官方参考
本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。
