Django WebSocket实时通信：Channels+Redis+异步视图

最近接了个内部工单系统的改造活儿需求一句话就能说完列表页有新工单进来不用刷新就得冒出来。听起来简单真动手才发现坑不少——Django 默认那套“请求-响应”的世界观天生不支持服务端主动推消息。折腾了两周最后落地的方案是 Django 的 WebSockets 加异步视图组合拳中间用 Channels 做协议桥接、Redis 做跨进程广播。这套组合跑下来单机撑个几千条长连接不算吃力代码量也比想象中少。这篇就把我踩过的路重新走一遍从为什么不用轮询、同步视图和异步视图该怎么分工、Consumer 怎么写才不会在鉴权上翻车一直到反向代理的超时配置和压测时冒出来的怪问题全部摊开讲。前端刚入门、只会写fetch的同学能看懂前面的选型部分后端同学可以直接抄后面的代码和参数。实时通信这词听着玄本质就是把“客户端问、服务端答”改成“服务端想起来就说”剩下的都是工程细节。1. 为什么要在 Django 里做实时通信需求拆解与技术选型1.1 从轮询到长连接几种方案的现实差距先摆事实。数据要实时到达前端主流就四条路短轮询、长轮询、SSE、WebSocket。短轮询最好写前端setInterval三秒一次后端加个接口就完事代价是九成的请求是空手而归。我算过一笔账500 个在线用户、3 秒轮询一次QPS 大概 167这里面可能只有 2 次真有数据。数据库和连接池被这些无效请求反复摩擦机器负载上去了用户看到的延迟还是最多 3 秒。长轮询是把请求挂住有数据或者超时才返回。它省掉了空请求但每个挂起的请求仍然占着一个服务端工作单元。Django 同步模式下一个请求占一个线程或进程挂 500 个就等于要 500 个并发槽位这在 WSGI 的世界里是灾难。SSE 走的是单向推送服务端到客户端实现简单、天然支持断线重连但它只能单工客户端想发消息还得另开一个 HTTP 接口。真正需要双向、低延迟、高频交互的场景WebSocket 才是正经答案一次 HTTP 升级握手之后连接就变成全双工两边随时可以发帧头部开销从几百字节降到几字节。判断标准很简单。如果你的场景是“通知类”比如工单提醒、订单状态变更、站内信客户端几乎不往回发东西SSE 完全够用运维成本还低。如果场景是“协作类”比如多人同时编辑、在线客服对话、实时光标位置同步、游戏房间双向高频是刚需那就上 WebSocket。我这次是工单系统但带在线客服模块所以直接选了 WebSocket省得以后再加一套。1.2 Django 同步视图与异步视图的边界在哪里很多人第一次碰 Django 异步会 confusion我写了async def的视图是不是整个项目就异步了不是。Django 从 3.1 开始支持异步视图但它的运行方式取决于你用什么服务器跑。用runserver或者任何 WSGI 服务器异步视图会被扔进一个线程里用asyncio.run()包起来执行等于花了异步的写法、付了同步的代价一点并发红利都吃不到。只有跑在 ASGI 服务器上async def视图才真正跑在事件循环里。那是不是所有视图都该改成异步我明确说不该。Django ORM 目前仍然是同步的你在异步视图里直接Model.objects.filter(...)会抛SynchronousOnlyOperation。要么用sync_to_async包一层那等于又回到线程池要么等 ORM 的异步接口铺开。Django 后来提供的aget、acreate这些异步 ORM 方法底层同样是在线程池里跑只是语法糖更顺手。我的分工原则是这样的短平快的 CRUD 接口保持同步简单、稳妥、生态兼容需要等待的、长连接的、IO 密集的入口用异步比如 WebSocket 消费者、聚合多个外部接口的报表页、需要await第三方 SDK 的接口。混着用完全没问题ASGI 服务器里同步视图会被自动调度到线程池你不用做任何额外配置。真正要注意的是别在异步代码里偷偷调同步阻塞函数比如requests.get()、time.sleep()、同步的文件读写这些会把整个事件循环卡住一个慢请求拖垮同进程所有连接。1.3 Channels 与 ASGI 的分工别把它们当成一个东西新手最容易搞混的就是 Channels 和 ASGI 的关系。ASGI 是一个协议规范定义了“服务器怎么和应用说话”它规定应用是一个async可调用对象接收scope、receive、send三个参数。Django 本身从 3.0 起就是个 ASGI 应用asgi.py里的get_asgi_application()返回的就是它。Channels 是在 ASGI 之上的一层框架它解决的是“Django 只认识 HTTP 请求”这个问题。它引入了三个概念消费者Consumer相当于 WebSocket 世界的视图函数路由routing把连接分发给对应的消费者通道层channel layer一个跨进程的消息队列抽象让 A 进程能通知 B 进程上的连接。没有通道层你的 WebSocket 连接绑死在某个进程上扩到多进程就彻底失效——用户在进程 1消息从进程 2 产生推不过去。所以架构上是这样的分层ASGI 服务器Uvicorn 或 Daphne负责监听和协议解析Channels 的ProtocolTypeRouter按协议把连接分流HTTP 交给 Django 原来的应用WebSocket 交给你的消费者。通道层用 Redis 这类外部存储做后端各大进程共享。理解了这个分层后面排查问题就有方向感了连不上是 ASGI 服务器或路由的问题连上了收不到消息是通道层或组名的问题收到消息但报错是消费者代码的问题。2. WebSocket 链路的关键细节握手、鉴权与消费者设计2.1 握手阶段的三件事路由、鉴权、作用域WebSocket 连接建立的第一步是一个 HTTP GET 请求带Upgrade: websocket头。服务器返回 101 就算切换协议成功之后就不再是 HTTP 了。这个握手阶段是你唯一能用 HTTP 那套机制做鉴权的窗口错过就没了。路由的写法是re_path匹配路径捕获的参数会进scope[url_route][kwargs]。这里有个很容易掉进去的坑路由顺序是从上往下匹配你的动态路径如果写在前面会把后面的静态路径全吃掉。我见过有人写re_path(rws/(?Proom\w)/$, ...)放在re_path(rws/health/$, ...)前面结果健康检查永远进不去因为health被当成 room 名匹配走了。鉴权方面Channels 提供了AuthMiddlewareStack它会把 HTTP 握手时的 session cookie 解析出来把用户塞进scope[user]。这对浏览器场景很方便但也有局限它依赖 session意味着跨域场景、移动端、第三方客户端都不好使因为这些客户端不一定带 cookie。另外浏览器原生的WebSocket构造函数不支持自定义请求头你没法像fetch那样加个Authorization: Bearer xxx。所以移动端或 token 方案只能用 URL query 参数传一次性票据这一点后面单独展开。作用域scope是贯穿整个连接生命周期的字典里面有path、headers、query_string、client客户端 IP 和端口、user还有subprotocols。我习惯在connect方法里把这些信息记一条结构化日志排查问题时特别有用——尤其是client里的 IP做限流和封禁全靠它。2.2 三种消费者形态与选择依据Channels 给了三个基类选错了会白写不少代码。AsyncWebsocketConsumer是最底层的你拿到的是原始文本或字节帧收发全靠await self.send(text_data...)。灵活但消息格式得自己管JSON 序列化反序列化、异常处理都要手写。JsonWebsocketConsumer同步版和AsyncJsonWebsocketConsumer异步版在它之上加了 JSON 编解码你收到的是已经json.loads过的 dict发的时候直接给 dict 就行。绝大多数业务场景用这个就够了。AsyncJsonWebsocketConsumer是我这次选的。理由是消息处理里要查库、要调外部接口异步写法能避免阻塞事件循环。如果你确定消费者里全是纯内存操作同步版也无所谓Channels 会自动把同步消费者调度到线程池但那样每个连接都占一个线程连接数一多线程池就打满了反而更糟。所以我的建议是只要用 WebSocket就默认走Async版本同步基类留给纯演示或极简场景。三个必须实现的方法connect握手时调用决定接受还是拒绝、disconnect断开时调用负责清理、receive_json收到消息时调用。还有一个隐式的xxx_message方法族用来处理从通道层推过来的消息这个机制是 Channels 的核心设计值得单独说。2.3 通道层的组播机制与几个致命细节通道层有两个基本动作send发给单个 channelgroup_send发给一个组。每个连接都有一个唯一的channel_name加入组之后任何进程往这个组group_send消息会广播给组内所有连接。消息到达客户端之前会经过消费者内部的一次分发你发的消息体里必须有一个type字段Channels 把它转成方法名。比如{type: chat.message, text: hi}会被路由到chat_message方法。点号变下划线这是硬规则写错了就是静默丢弃日志里啥也没有我第一次踩这个坑找了两个小时。还有两个参数必须理解。capacity是每个 channel 的队列长度默认 100超过之后新消息会被丢掉。expiry是消息在队列里的存活秒数默认 60超时未消费也丢。这两个参数决定了通道层的“最多一次”投递语义——它不保证送达只保证尽力。所以别把关键业务数据只压在 WebSocket 上数据库落库才是真相来源推送只是加速体验。我的做法是先写库事务提交后再推消息前端收到推送只做 UI 更新真要对账就重新拉接口。这样即使推送丢了用户刷新一下也就回来了。组名也有讲究它全局共享一个命名空间。如果你用group_add(fuser_{id})得保证不同业务线不会撞名我习惯加前缀比如notify_user_{id}、chat_room_{id}。另外组的成员关系是存在 Redis 里的进程崩溃时不会自动清理得靠expiry兜底Channels 会定期扫描过期的组成员。3. 实操从零搭一条可用的实时通知通道3.1 环境依赖与 ASGI 配置先把依赖列清楚。我用的组合是 Django 4.2 Channels 4.x channels-redis 4.x Redis 7 Uvicorn。版本上有一点要强调Channels 4 把channels_redis的包名从channels_redis改成了channels_redis对名字没变但内部 API 变了老教程里的asgi_redis早就废弃了。如果你的 Django 是 3.xChannels 3.x 更稳。pip install django4.2 channels[daphne]4.0 channels-redis4.0 uvicorn装完在INSTALLED_APPS里把daphne放到django.contrib.staticfiles前面如果你用 Daphne 做开发服务器这样runserver会自动切换成 ASGI 模式。我用 Uvicorn所以这步可以省但为了runserver的便利还是加上了。settings.py里两个关键配置ASGI_APPLICATION myproject.asgi.application CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], capacity: 1500, expiry: 30, group_expiry: 86400, }, }, }capacity我从默认的 100 调到了 1500原因是慢消费者问题某个客户端网络差消息堆在它自己的 channel 队列里队列一满group_send就会抛ChannelFull异常。如果这个异常冒到广播方会连累其他正常用户的推送。调大容量能缓冲一阵但根治还得靠前端消费速度和后端的降级策略。expiry设 30 秒是为了避免陈旧消息在断线重连后被补发用户看到一堆过时通知不是什么好体验。asgi.py的写法import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.security.websocket import AllowedHostsOriginValidator from channels.auth import AuthMiddlewareStack from notify.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: AllowedHostsOriginValidator( AuthMiddlewareStack(URLRouter(websocket_urlpatterns)) ), })AllowedHostsOriginValidator这层千万别省。WebSocket 有个叫跨站 WebSocket 劫持的问题攻击者在自己域名下开个页面让你的用户浏览器连你的服务器浏览器会自动带上 session cookie于是攻击者就拿到了一个已认证的连接。这个校验器会检查Origin头不在ALLOWED_HOSTS里的直接拒绝。上线前我专门用 curl 模拟了一次伪造 Origin被 403 挡住才算放心。3.2 消费者代码一个能跑的通知通道先看路由from django.urls import re_path from . import consumers websocket_urlpatterns [ re_path(r^ws/notify/$, consumers.NotifyConsumer.as_asgi()), ]消费者本体import json from channels.generic.websocket import AsyncJsonWebsocketConsumer from channels.db import database_sync_to_async class NotifyConsumer(AsyncJsonWebsocketConsumer): async def connect(self): user self.scope.get(user) if not user or not user.is_authenticated: # 拒绝握手前端会收到 403 await self.close(code4401) return self.group_name fnotify_user_{user.id} await self.channel_layer.group_add(self.group_name, self.channel_name) await self.accept() unread await self.get_unread_count(user.id) await self.send_json({type: connected, unread: unread}) async def disconnect(self, code): if hasattr(self, group_name): await self.channel_layer.group_discard( self.group_name, self.channel_name ) async def receive_json(self, content, **kwargs): action content.get(action) if action ping: await self.send_json({type: pong, ts: content.get(ts)}) elif action read: await self.mark_read(content.get(ids, [])) async def notify_message(self, event): # 对应 group_send 里的 type: notify.message await self.send_json({ type: notification, title: event[title], body: event[body], ts: event[ts], }) database_sync_to_async def get_unread_count(self, user_id): from .models import Notification return Notification.objects.filter(user_iduser_id, readFalse).count() database_sync_to_async def mark_read(self, ids): from .models import Notification if ids: Notification.objects.filter(id__inids).update(readTrue)几个细节值得掰开。close(code4401)用的是自定义关闭码4000-4999 是应用层可以自定义的区间前端拿到 4401 就知道是“未登录”不要去重连而是跳登录页4402 可以定义为“被踢下线”同样不重连。这一点比在连接建立后再发一条“你未登录”的消息要干净得多省了一次无谓的握手。database_sync_to_async是 Channels 提供的装饰器它比 Django 的sync_to_async多做了连接管理——同步 ORM 在线程里跑完会自动关闭数据库连接避免连接泄漏。这个装饰器必须用别用错成sync_to_async长期跑下来数据库连接数会失控。notify_message这个方法的命名必须和group_send里的type对应notify.message转成notify_message。我一开始写成notifyMessage死活收不到后来才反应过来规则。3.3 业务代码里怎么触发推送推送入口一般落在同步视图或 Celery 任务里同步代码加异步通道层需要async_to_sync包一层from asgiref.sync import async_to_sync from channels.layers import get_channel_layer from django.db import transaction def push_notification(user_id, title, body): channel_layer get_channel_layer() async_to_sync(channel_layer.group_send)( fnotify_user_{user_id}, { type: notify.message, title: title, body: body, ts: timezone.now().isoformat(), }, )这里必须加事务保护这是一个我摔过跟头的地方with transaction.atomic(): n Notification.objects.create(user_iduser_id, titletitle, bodybody) transaction.on_commit(lambda: push_notification(user_id, title, body))为什么要on_commit因为group_send是立即执行的如果它在事务提交之前就推到了前端而前端收到消息后立刻回一个“标记已读”的请求这个请求走的是另一个数据库连接看不到未提交的数据会报 404 或者静默失败。更糟的是如果事务最终回滚用户已经收到了一条根本不存在的通知。on_commit保证推送只在事务真正落库后才发生顺序是对的。3.4 前端连接、心跳与重连浏览器原生 WebSocket 没有自动重连也没有心跳机制JS 没法发 ping 帧那是协议层的所以应用层心跳是必须自己写的。别指望 TCP 层的 keepalive 能救你运营商网关和负载均衡器经常在 60 到 300 秒内掐掉空闲连接而且掐得无声无息——前端不报错只是再也收不到消息。class NotifySocket { constructor(url) { this.url url; this.retry 0; this.maxRetry 10; this.heartbeatTimer null; this.lastPong Date.now(); this.connect(); } connect() { this.ws new WebSocket(this.url); this.ws.onopen () { this.retry 0; this.lastPong Date.now(); this.startHeartbeat(); }; this.ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type pong) { this.lastPong Date.now(); return; } // 业务分发 this.onNotify this.onNotify(msg); }; this.ws.onclose (e) { this.stopHeartbeat(); if (e.code 4401) { window.location.href /login/; return; } if (e.code 4402) return; this.scheduleReconnect(); }; this.ws.onerror () this.ws.close(); } startHeartbeat() { this.heartbeatTimer setInterval(() { if (this.ws.readyState ! WebSocket.OPEN) return; if (Date.now() - this.lastPong 45000) { this.ws.close(); return; } this.ws.send(JSON.stringify({ action: ping, ts: Date.now() })); }, 25000); } stopHeartbeat() { clearInterval(this.heartbeatTimer); } scheduleReconnect() { if (this.retry this.maxRetry) return; const delay Math.min(1000 * Math.pow(2, this.retry), 30000); const jitter Math.random() * 1000; this.retry 1; setTimeout(() this.connect(), delay jitter); } }心跳间隔 25 秒、判定超时 45 秒这组数字不是拍脑袋的。绝大多数中间设备的空闲超时在 60 秒以上25 秒发一次能稳定地“喂”住它们。判定超时设成接近两倍间隔是为了容忍一次丢包。重连用指数退避加随机抖动抖动很重要——如果服务端重启几千个客户端会在同一毫秒同时重连形成惊群。加了 0 到 1 秒的随机抖动后重连请求就被摊平了。3.5 Redis 通道层的容量估算容量参数不是随便填的得按峰值算。我整理了一个估算表直接套公式就行参数含义估算依据我的取值capacity单 channel 队列长度峰值推送速率 × 可容忍积压秒数1500expiry消息队列存活时间前端心跳周期 × 1.5 左右30 秒group_expiry组成员过期清理单次最长连接时长上浮86400 秒Redis 内存通道层占用连接数 × 平均队列长度 × 单条字节数约 300MB具体算一遍峰值在线 3000 连接每连接平均每秒收 0.2 条消息正常情况下队列是空的。极端情况是某个客户端网络卡住 10 秒积压 2 条容量 1500 覆盖得绰绰有余。内存这边3000 个 channel每个 channel 自带一些元数据大约 100 到 200 字节加上组映射总占用在百兆级别。如果你的连接数上万建议把 Redis 单独部署到一个实例别和缓存混用通道层的写入模式频繁的LPUSH/BRPOP和普通缓存的读密集模式差别很大混在一起容易互相影响。4. 认证与权限把匿名连接挡在门外4.1 会话 Cookie 鉴权的边界在哪里AuthMiddlewareStack靠 cookie 认人浏览器同源场景下这是最省事的方案连前端代码都不用改。但它有个隐含前提你的页面和 WebSocket 端点必须同源或者至少在ALLOWED_HOSTS和CSRF_TRUSTED_ORIGINS里互相认识。跨子域的时候cookie 的domain属性得设对Secure和SameSite也要跟着配SameSiteLax在跨站场景下不会带 cookie握手就会被当成匿名用户。还有一个更隐蔽的坑session 的过期。HTTP 请求里 session 过期用户被重定向到登录页很直观。但 WebSocket 连接一旦建立session 就不再被检查了——用户可以一直连着会话早就过期了他还能继续收推送。这个问题在弱网环境下特别明显用户合上笔记本第二天打开连接可能还活着或者刚重连成功但会话已经失效。我的处理是在connect里做一次主动校验并且给连接设一个最长存活时间。校验逻辑不复杂拿到scope[user]检查user.is_authenticated再检查 session 的过期时间是否已过。定期把连接踢掉让前端重新握手代价是每分钟几千次重连收益是安全性可控。4.2 一次性票据方案给非浏览器客户端留门移动端、桌面客户端、第三方集成这些都不带 cookie浏览器原生 WebSocket 又不支持自定义头。唯一的干净出路是“短票据换连接”客户端先用正常的 HTTP 接口带 token 认证申请一张一次性的、有效期极短的票据然后拿着这张票据去连 WebSocketURL 上带?ticketxxx。import secrets from django.core.cache import cache def issue_ticket(user_id): ticket secrets.token_urlsafe(32) cache.set(fws_ticket:{ticket}, user_id, timeout30) return ticket在自定义中间件里消费这张票据from channels.db import database_sync_to_async from django.contrib.auth import get_user_model from urllib.parse import parse_qs database_sync_to_async def get_user(user_id): return get_user_model().objects.filter(iduser_id).first() class TicketAuthMiddleware: def __init__(self, inner): self.inner inner async def __call__(self, scope, receive, send): query parse_qs(scope.get(query_string, b).decode()) ticket (query.get(ticket) or [None])[0] if ticket: user_id cache.get(fws_ticket:{ticket}) if user_id: cache.delete(fws_ticket:{ticket}) # 一次性用完即焚 scope[user] await get_user(user_id) return await self.inner(scope, receive, send)注意cache.delete那行票据必须一次性。如果不删票据在 30 秒有效期内可以被反复使用一旦泄露就等于永久后门。token_urlsafe(32)生成的是 256 位熵的随机串暴力猜解不现实。这个方案有个取舍票据走的是 query string而 query string 会进 nginx 的 access log、会进浏览器的历史记录浏览器场景下。所以浏览器场景还是用 cookie 方案票据方案专门留给不自带 cookie 的客户端两者不要混。日志这边我在 nginx 配置里对ws_ticket参数做了脱敏替换虽然麻烦但总比票据躺在日志里强。4.3 连接数与资源限制认证挡住了匿名还得挡住恶意刷连接。一个已登录用户理论上可以开无数个 WebSocket每个连接在服务端都有内存开销几千个连接就能把单机拖垮。限制手段分三层。应用层在connect里做计数用 Redis 的INCR给每个用户维护一个活跃连接数超过阈值我设的 5就拒绝断线时DECR同时给 key 设过期时间兜底防止进程崩溃导致计数永远不降。代理层在 nginx 里按 IP 限并发limit_conn模块几行配置就能拦住单 IP 的洪水连接。系统层调ulimit -n默认 1024 个文件描述符几千连接必然不够改成 65535 是标配。顺便说一句拒绝连接的时候不要用 HTTP 层的 429WebSocket 握手阶段直接close(code4429)更清晰前端可以在 UI 上提示“连接数过多请关闭其他标签页”。5. 部署与运维ASGI 服务器、反向代理与压测5.1 进程模型与 worker 数量估算WSGI 的经典公式是 workers 2 × CPU 核数 1但那个公式是为“请求即来即走”的模型设计的套在 WebSocket 上会翻车。原因是长连接常驻每个连接在事件循环里挂着一个协程协程本身很轻几 KB但 worker 崩溃时上面的所有连接会同时断开。worker 越多单个 worker 挂掉影响面越小但内存占用也越高。我的算法分两步。先算连接密度在目标机器上压测观察单个 worker 能稳挂多少连接我的实测是 4 核 8G 的机器上单个 Uvicorn worker 挂 2000 个空闲连接时CPU 占用在 10% 以下内存增加约 300MB。再按业务峰值除以这个密度得到 worker 数。3000 连接的话2 个 worker 就够留一个余量取 3。uvicorn myproject.asgi:application \ --host 0.0.0.0 --port 8000 \ --workers 3 \ --loop uvloop \ --ws websockets \ --ws-ping-interval 20 \ --ws-ping-timeout 20 \ --proxy-headers--ws websockets指定用websockets库做协议实现比默认的实现性能好一些。--ws-ping-interval是协议层的 ping 帧如果你在应用层已经做了心跳这里可以关掉设成 0两层心跳叠加纯属浪费。--proxy-headers一定要加否则反向代理后面的客户端 IP 全变成 127.0.0.1限流和日志全废。前提是你配了FORWARDED_ALLOW_IPS或者信任代理的 IP 段不然这个头可以被客户端伪造。5.2 反向代理的几个必改项nginx 默认配置用来跑 WebSocket 是一定会断的问题出在三个地方。location /ws/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_read_timeout 3600s; proxy_send_timeout 3600s; proxy_buffering off; }proxy_http_version 1.1和那两个Upgrade/Connection头是协议升级的必需项缺一个握手就失败nginx 会把请求当普通 HTTP 转发给后端后端一脸懵。proxy_read_timeout默认 60 秒到点就掐连接这是“连上后一分钟必断”的最常见原因改成 3600 秒。proxy_buffering off是为了让帧实时透传开着缓冲的话消息会攒着一起发实时性就没了。还有一个容易忽略的如果前面还有一层云厂商的负载均衡它们通常有自己的空闲超时常见 60 秒、也有 900 秒的这个超时你改不了只能靠心跳去喂。心跳间隔必须小于这个值这就是前面为什么选 25 秒——它要同时压过 nginx 和应用层的所有超时。5.3 监控指标与压测方法没压测过的实时功能等于没做。压测脚本我用asyncio加websockets库手写因为现成的压测工具大多按 HTTP 思路设计对长连接的支持一般。import asyncio, time, statistics import websockets async def one_client(url, hold, results): t0 time.perf_counter() try: async with websockets.connect(url, extra_headers{Cookie: COOKIE}) as ws: lat time.perf_counter() - t0 await asyncio.sleep(hold) results.append(lat) except Exception as e: results.append(-1) async def main(n, hold): results [] tasks [one_client(URL, hold, results) for _ in range(n)] await asyncio.gather(*tasks) ok [r for r in results if r 0] print(f成功 {len(ok)}/{n}, 握手 P95 {statistics.quantiles(ok, n20)[18]:.3f}s) asyncio.run(main(2000, 120))要看的指标有四个握手延迟 P95反映鉴权和路由的开销正常应该低于 200ms、连接保持成功率两分钟内不掉线目标是 99% 以上、广播延迟从触发推送到客户端收到的端到端时间参考值 500ms 内、进程内存增长连接稳定后应该走平持续上涨说明有泄漏。我在一次压测里发现内存以每分钟 50MB 的速度上涨排查下来是disconnect里group_discard没执行——因为我把self.group_name的赋值放在了accept()之后而某些异常路径下accept之前就断开了hasattr判断失效。后来改成在connect最开头就赋值问题消失。还有一个常见指标是ChannelFull异常的计数。这个异常一般不抛到业务层因为group_send内部会 catch 掉但日志里会有警告。如果这个数字持续上涨说明有慢消费者得去看前端是不是有页面标签被挂起了、或者某个客户端在弱网下疯狂积压。6. 常见问题与排查实录6.1 故障速查表现象最可能原因排查动作解决握手直接 403Origin 不在白名单看AllowedHostsOriginValidator日志补ALLOWED_HOSTS或CSRF_TRUSTED_ORIGINS连上 60 秒必断nginxproxy_read_timeout抓包看断开时机调到 3600s单机能通多进程不通没用通道层或 Redis 连错库看CHANNEL_LAYERS配置统一 Redis检查 db 编号消息发出收不到type字段和方法名不匹配打印消费者收到的 event点号转下划线命名对齐间歇ChannelFull慢消费者积压统计异常次数和 channel 分布调大 capacity前端加快消费用户 A 收到用户 B 的消息组名没带用户 ID查group_send的组名构造组名加唯一标识数据库连接数暴涨用了sync_to_async而非database_sync_to_async看连接池监控换成 Channels 提供的装饰器重连风暴打挂后端没加退避和抖动看服务端 QPS 曲线指数退避 随机抖动6.2 几个只有踩过才知道的点别在消费者里做重活。事件循环是单线程的你在receive_json里跑一个 500ms 的同步计算同进程的所有连接这 500ms 全都僵住。要用asyncio.to_thread或者database_sync_to_async甩出去。我做过对比一个 200ms 的同步阻塞放进消费者2000 连接的场景下整体延迟从 30ms 飙到 400ms非常直观。跨进程广播必须靠通道层没有捷径。有人试过用全局字典存连接对象单进程跑得好好的一上多 worker 就失灵。这不是 bug是进程隔离的必然结果。如果你暂时不想上 Redis就用单 worker 跑Uvicorn 的--workers 1但要清楚这是有上限的扛不住就得上通道层。事务和推送的顺序错了会很难查。前面提过on_commit这里补一句Celery 任务里推送不要紧因为任务已经在事务提交后跑了但视图里直接推一定要包在on_commit里。这个错误的表现是“偶发的前端 404”很难复现因为取决于数据库的事务隔离级别和并发时序。关闭码是前后端的契约。4000 到 4999 是你自己的地盘定义好文档4401 未登录、4402 被顶替、4429 超限、4500 服务端内部错误。前端按码分流该跳登录的跳登录该重连的重连该弹提示的弹提示。我见过把所有关闭都当成“重连”处理的实现结果是用户被踢了之后疯狂重连服务端日志被刷爆。日志要带连接标识。channel_name是一串随机字符但它能把同一个连接的握手、消息、断开、异常全串起来。我在每条日志里都带上它的前 8 位排查时 grep 一下整条时间线就出来了比什么都好用。我在实际项目里最大的体会是实时功能的复杂度不在“连上”而在“连不上之后怎么办”。断线、重连、消息补发、多端登录冲突、会话过期这些边角情况占了整个工作量的七成。真正跑起来之后我建议先别急着优化性能先把断线重连的每一种分支都手动模拟一遍——拔网线、切 WiFi、休眠唤醒、后端重启、主动踢人五个场景全部走通才算这套东西真的能用。