企业官网建设流程全解析

最近接了个内部工单系统的改造活儿，需求一句话就能说完：列表页有新工单进来，不用刷新就得冒出来。听起来简单，真动手才发现坑不少——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(r"ws/(?P<room>\w+)/$", ...)放在re_path(r"ws/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发给单个 channel，group_send发给一个组。每个连接都有一个唯一的channel_name，加入组之后，任何进程往这个组group_send，消息会广播给组内所有连接。

消息到达客户端之前，会经过消费者内部的一次分发：你发的消息体里必须有一个"type"字段，Channels 把它转成方法名。比如{"type": "chat.message", "text": "hi"}，会被路由到chat_message方法。点号变下划线，这是硬规则，写错了就是静默丢弃，日志里啥也没有，我第一次踩这个坑找了两个小时。

还有两个参数必须理解。capacity是每个 channel 的队列长度，默认 100，超过之后新消息会被丢掉。expiry是消息在队列里的存活秒数，默认 60，超时未消费也丢。这两个参数决定了通道层的“最多一次”投递语义——它不保证送达，只保证尽力。所以别把关键业务数据只压在 WebSocket 上，数据库落库才是真相来源，推送只是加速体验。我的做法是：先写库，事务提交后再推消息，前端收到推送只做 UI 更新，真要对账就重新拉接口。这样即使推送丢了，用户刷新一下也就回来了。

组名也有讲究，它全局共享一个命名空间。如果你用group_add(f"user_{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.x，Channels 3.x 更稳。

pip install "django>=4.2" "channels[daphne]>=4.0" "channels-redis>=4.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(code=4401) return self.group_name = f"notify_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_id=user_id, read=False).count() @database_sync_to_async def mark_read(self, ids): from .models import Notification if ids: Notification.objects.filter(id__in=ids).update(read=True)

几个细节值得掰开。close(code=4401)用的是自定义关闭码，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)( f"notify_user_{user_id}", { "type": "notify.message", "title": title, "body": body, "ts": timezone.now().isoformat(), }, )

这里必须加事务保护，这是一个我摔过跟头的地方：

with transaction.atomic(): n = Notification.objects.create(user_id=user_id, title=title, body=body) 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 通道层的容量估算

容量参数不是随便填的，得按峰值算。我整理了一个估算表，直接套公式就行：

具体算一遍：峰值在线 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也要跟着配，SameSite=Lax在跨站场景下不会带 cookie，握手就会被当成匿名用户。

还有一个更隐蔽的坑：session 的过期。HTTP 请求里 session 过期，用户被重定向到登录页，很直观。但 WebSocket 连接一旦建立，session 就不再被检查了——用户可以一直连着，会话早就过期了，他还能继续收推送。这个问题在弱网环境下特别明显，用户合上笔记本第二天打开，连接可能还活着，或者刚重连成功，但会话已经失效。

我的处理是在connect里做一次主动校验，并且给连接设一个最长存活时间。校验逻辑不复杂：拿到scope["user"]，检查user.is_authenticated，再检查 session 的过期时间是否已过。定期把连接踢掉让前端重新握手，代价是每分钟几千次重连，收益是安全性可控。

4.2 一次性票据方案：给非浏览器客户端留门

移动端、桌面客户端、第三方集成，这些都不带 cookie，浏览器原生 WebSocket 又不支持自定义头。唯一的干净出路是“短票据换连接”：客户端先用正常的 HTTP 接口（带 token 认证）申请一张一次性的、有效期极短的票据，然后拿着这张票据去连 WebSocket，URL 上带?ticket=xxx。

import secrets from django.core.cache import cache def issue_ticket(user_id): ticket = secrets.token_urlsafe(32) cache.set(f"ws_ticket:{ticket}", user_id, timeout=30) 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(id=user_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(f"ws_ticket:{ticket}") if user_id: cache.delete(f"ws_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 层的 429，WebSocket 握手阶段直接close(code=4429)更清晰，前端可以在 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, n=20)[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 故障速查表

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、休眠唤醒、后端重启、主动踢人，五个场景全部走通，才算这套东西真的能用。