前阵子公司内部一个导出报表的任务动不动就跑二三十分钟。之前是前端每 5 秒轮询一次后端任务状态用户盯着转圈圈的时候你看到的其实是“上一次轮询时还不知道是不是活着的数据”。这逼着我把整条链路重做了一遍后端状态追踪 WebSocket 实时推送 前端页面展示。做完以后不仅任务进度秒级可见排障的时候也能直接看到消息流。这篇文章适合后端为主、前端够用、想一条龙打通实时状态展示的团队尤其是任务编排、构建系统、文件批量处理这类场景。里面涉及的状态设计、心跳、断线补偿、反向代理配置都是我实际跑过线上以后整理出来的可以直接抄。1. 先把后端状态机画明白再谈推送不迟很多人一听到“后端状态追踪 WebSocket 实时推送”第一反应就是赶紧去写 WebSocket 服务端。实际上我在项目里踩的第一个坑不是在 WebSocket 本身而是后端的状态字段设计得一团乱。状态没定义清楚推送出去的消息自然也是乱的前端收到以后只能在 switch-case 里打补丁。1.1 用数字当状态码是第一个坑老项目里有个 task 表status 字段是0/1/2当时注释写着 0 待处理、1 处理中、2 已完成。前期只有三个状态的时候确实简单SQL 里update task set status1写起来也爽。但后来需求加了“排队中”“取消中”“重试中”问题就全冒出来了有的地方用数字、有的地方用字符串做判断前后端各维护一份映射表只要有一边没同步页面就出现“未知状态”。我的建议是状态直接使用语义化字符串或者后端定义 Python Enum然后再序列化成字符串输出。比如pending / running / success / failed / cancelled前端收到的是可读的状态名展示文案由前端映射。哪怕只是多几个字符也不要为了“省流量”去用数字。因为 WebSocket 推送的主要成本是连接维护不是这几个字节。1.2 状态迁移约束不是所有状态都能随便跳定义好状态值只是第一步更重要的是“谁允许跳到谁”。任务从pending进入running没问题但从success跳回running就是严重 bug。这类问题如果散落在业务代码里每次更新状态都得自己判断一遍早晚会漏。我在服务端统一封装了一个transition_to(new_state)方法所有状态变更都必须走这个方法。内部维护一张迁移表当前状态允许迁移到的状态pendingrunning / cancelledrunningsuccess / failed / cancelledcancelled无终态success / failed无终态实际项目里还可以允许failed - pending做重试这取决于业务。关键是“迁移规则只有一份”而不是每个写任务的地方都来一套 if-else。这样后端状态机稳定后续做 WebSocket 推送时每次transition_to成功之后顺手发一条事件消息不会漏推。1.3 消息体设计type、payload、timestamp、traceId状态机定完接下来是推送消息本身。新手最容易犯的错是把整个 task 对象直接 JSON 序列化推给前端。这种方式第一次看没什么问题可一旦任务对象里塞进日志、内部字段前端就会拿到一堆用不上的数据而且类型很泛不好处理。我用的消息结构长这样{ type: task.progress, payload: { taskId: 8f0c2f3e-..., status: running, progress: 47, message: 正在导入 2024 年销售数据 }, timestamp: 1712400001234, traceId: trace-8f0c2f3e-01 }type 是事件类型比如task.created、task.progress、task.completedpayload 是业务数据timestamp 是服务端产生时间前端可以用它判断消息是否过期traceId 贯穿后端日志和推送链路排查问题的时候特别有用。前端只需要根据 type 做分发payload 里的字段在 TypeScript 里定义好对应类型即可。有人可能会问为什么不用 status 当 type因为业务事件不止状态变化还有“阶段切换”“进度更新”“取消确认”等。状态是最终展示结果事件是发生了什么两者分开前端处理起来会清晰得多。2. 后端推送链路Django Channels 连接管理与跨进程消息后端我用的是 Django Channels。选择它的原因很直接项目已经建立在 Django 上需要复用鉴权和 ORM而且实时推送只是其中一环不值得为这个单独引入一套新框架。如果你的项目不是 Django或者只需要一条极简推送通道FastAPI Redis pub/sub 反而更轻。这里不拉踩只讲清楚场景。2.1 选型什么情况下选 Channels什么情况不如自己写Channels 把 WebSocket 的接入层包装成了类似 Django view 的消费者consumer同时提供 channel layer 做跨进程通信。它适合项目已经在用 Django并且 WebSocket 连接需要和登录体系打通需要把消息广播到多个客户端比如任务组、聊天室后台任务Celery 等和 WebSocket 服务不在同一个进程里你不想自己维护连接状态和进程间消息路由。如果只是做一个内部小工具后端不碰 Django一个websockets库加 Redis 就足够了。我当时没有重新造轮子是因为任务系统里有一堆现有的 model 和权限逻辑用 Channels 可以把“任务状态变更”和“实时推送”两个功能写在一个事务上下文里维护成本最低。2.2 路由与消费者连接进来之后发生什么先看 ASGI 配置# config/asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from django.urls import path from tasks.consumers import TaskConsumer os.environ.setdefault(DJANGO_SETTINGS_MODULE, config.settings) application ProtocolTypeRouter({ http: get_asgi_application(), websocket: URLRouter([ path(ws/tasks/uuid:task_id/, TaskConsumer.as_asgi()), ]), })TaskConsumer 的核心结构# tasks/consumers.py import json from channels.generic.websocket import AsyncWebsocketConsumer from channels.db import database_sync_to_async class TaskConsumer(AsyncWebsocketConsumer): async def connect(self): self.task_id self.scope[url_route][kwargs][task_id] self.group_name ftask_{self.task_id} # 鉴权、业务校验省略伪代码 user self.scope.get(user) if not user or not await self.can_access_task(user, self.task_id): await self.close(code4403) return # 加入频道组便于按任务维度推送 await self.channel_layer.group_add(self.group_name, self.channel_name) await self.accept() await self.send_json({type: task.connected, payload: {...}}) async def disconnect(self, code): await self.channel_layer.group_discard(self.group_name, self.channel_name) async def receive(self, text_dataNone, bytes_dataNone): data json.loads(text_data) if data.get(type) heartbeat: await self.send_json({type: heartbeat_ack, timestamp: ...}) async def task_push(self, event): # 注意group_send 里的 type 会映射到 consumer 的方法 task_push await self.send_json(event[event])这里有个关键点group_add把当前连接加入了以task_id命名的频道组。后文无论哪个进程执行group_send(task_id, {...})Channels 的 channel layer 都会把消息路由到持有这个连接的消费者实例上。这也解释了为什么多进程部署下依然能准确推送——只要所有 worker 连接的是同一个 Redis。2.3 鉴权放在握手阶段WebSocket 没有 View 中间件WebSocket 的鉴权和普通 HTTP 请求不太一样。浏览器发送 WebSocket 握手请求时可以带 Cookie也可以带查询参数但处理方式更像一个长连接的“连接鉴权”。如果连接通过了后续消息再逐条鉴权会非常麻烦。我在生产环境里的做法是前端调用普通登录接口拿到短期 token然后在new WebSocket(url ?token token)时带上。服务端在 connect 阶段解析 query string再做一次鉴权。由于通道连接建立后就不能通过 HTTP Header 动态改权限所以这种“连接时鉴权”是主流做法。需要注意token 放 URL 有一个潜在风险Nginx 默认会把完整 URL 写到 access logtoken 会被记录下来。解决思路有两个一是让 token 非常短时比如 10 分钟有效且只用于握手二是重写日志格式把查询参数脱敏。我实际用的是短 token即使泄漏攻击者也只能在失效前建立一条 WebSocket 连接危害可控。2.4 后台任务怎么把状态塞进 WebSocket跨进程通道这是最容易懵的地方。很多同学以为在 Celery 任务里直接调用 WebSocket 的 send 就行但 Celery worker 进程和 Daphne/Uvicorn 进程根本不是同一个进程拿不到连接对象。Channels 的 channel layer 就是为解决这个问题设计的。后台任务里这样推送# tasks/jobs.py from channels.layers import get_channel_layer from asgiref.sync import async_to_sync def send_task_event(task_id: str, event: dict): channel_layer get_channel_layer() async_to_sync(channel_layer.group_send)( ftask_{task_id}, { type: task.push, event: event, }, )如果任务本身是异步代码直接await channel_layer.group_send(...)就可以。group_send 消息里的type字段很关键Channels 会根据它调用消费者上同名方法所以task.push对应消费者里的async def task_push。这里有一个实战顺序问题一定要在数据库事务提交之后再推送消息。否则前端收到task.completed后立刻调用 REST 接口查询任务详情可能因为事务未提交而查到旧状态。我在项目里踩过表现为“页面已经显示完成刷新后又是处理中”看着非常精神分裂。正确做法是先提交任务状态的变更再发推送如果推送失败前端还能通过 REST 查询兜底。2.5 心跳机制TCP 不会主动告诉你客户端断线了WebSocket 建立在 TCP 之上而 TCP 有一个特性当客户端拔了网线、手机切换 Wi-Fi、或者电脑从休眠唤醒时网络层可能不会立刻通知服务端连接已失效。服务端看这条连接还是 ESTABLISHED实际已经收不到任何数据了。如果不处理服务端会留下一堆“幽灵连接”频道组里也会累积大量无效 channel。这就是为什么心跳机制几乎是 WebSocket 项目必备。我采用的方案是客户端每 30 秒发送一个 JSON 心跳包{type: heartbeat}服务端在 receive 里更新self.last_seen同时启动一个后台任务定时检查import asyncio, time async def check_alive_loop(self): while True: await asyncio.sleep(30) if time.time() - self.last_seen 90: await self.close(code4400) # 主动关闭僵尸连接 break这个方法实现简单而且心跳包本身就能在 WebSocket 调试面板里看到出问题时很容易排查。如果你追求更省流量可以用 WebSocket 协议层的 ping/pong 控制帧但那需要处理底层 ASGI 消息调试起来还不如 JSON 直观。对于任务状态推送这种低频场景JSON 心跳完全够用。3. 前端展示从轮询切换到 WebSocket 的完整封装思路后端推得再稳前端不会接也是白搭。React 项目里最常见的做法是直接在组件里new WebSocket()然后 onmessage 里 setState。Demo 能跑但线上撑不住组件卸载没关连接、断线不重连、消息频率一高页面渲染卡顿。下面是我总结的一套封装思路。3.1 轮询、SSE、WebSocket 到底怎么选先把方案对比摆出来方案延迟连接方向自动重连典型场景轮询取决于间隔秒级单向请求天然低频状态、后端改动成本低SSE秒级以内单向服务端到客户端浏览器内置通知、日志流、文件变化WebSocket实时双向需要自己写实时协作、任务控制、聊天、行情如果你的需求只是“服务端有文件变化就往页面推”SSE 其实更简单EventSource 自带重连后端也不用处理连接池。但如果后续要支持“用户在页面上取消任务”“手动触发重跑”就一定要 WebSocket因为它是双向的。React SSE/WebSocket 做文件变化推送我也写过本质就是用长连接替代轮询文件系统状态前端把事件类型和文件路径放进列表再按路径 debounce。高频场景下重点是前端合并事件这个我在 3.3 节单独讲。3.2 封装 useTaskWebSocket连接、收消息、自动重连我把连接逻辑封装成一个 Hook避免每个组件各自维护一套 WebSocket 生命周期。核心逻辑如下function useTaskWebSocket(taskId, { enabled true, onEvent }) { const onEventRef useRef(onEvent); onEventRef.current onEvent; useEffect(() { if (!enabled || !taskId) return; let disposed false; let ws null; let retries 0; const connect () { if (disposed) return; ws new WebSocket( ${location.protocol https: ? wss : ws}://${location.host} /ws/tasks/${taskId}/?token${encodeURIComponent(getToken())} ); ws.onopen () { retries 0; }; ws.onmessage (e) { const msg JSON.parse(e.data); if (msg.type heartbeat_ack) return; onEventRef.current(msg); }; ws.onclose () { if (disposed) return; const delay Math.min(30000, 1000 * 2 ** retries) Math.random() * 1000; retries 1; setTimeout(connect, delay); }; }; connect(); return () { disposed true; ws ws.close(); }; }, [taskId, enabled]); }几个细节解释一下用onEventRef保存最新的回调避免回调函数变化导致 effect 反复执行。disposed标记防止组件卸载后定时器触发重连。重连间隔使用指数退避最多 30 秒加一点随机抖动防止多个客户端同时重连形成惊群。清理函数里一定要ws.close()否则组件卸载后连接还挂在服务端。如果你用 React StrictModeeffect 会执行两次这个清理逻辑必须写。否则开发模式下会出现两条 WebSocket 连接后端日志里莫名多了断连记录又得浪费半小时。3.3 高频推送下的渲染优化别让状态更新打爆页面任务进度如果后端每秒推 10 条前端每收到一条就setProgressReact 会频繁渲染页面会明显卡顿。解决办法有两个方向一是让后端做节流但节流会牺牲实时性二是前端做消息合并。我的做法是把最后一条有效消息先存到 ref然后用requestAnimationFrame统一刷新。const latestEventRef useRef(null); const rafIdRef useRef(null); const handleEvent (msg) { latestEventRef.current msg; if (rafIdRef.current ! null) return; rafIdRef.current requestAnimationFrame(() { rafIdRef.current null; const event latestEventRef.current; latestEventRef.current null; if (!event) return; // 这里统一 setState applyTaskEvent(event); }); };这样做的好处是即使一帧内收到 20 条消息也只会触发一次渲染渲染的是最新状态。对进度条这种东西来说“跳过中间值”完全没问题用户反而会觉得更流畅。还有一个容易被忽略的点后端可能重复推送相同状态前端在applyTaskEvent里先判断 progress / status 是否变化没变化就不更新。3.4 断线恢复后的状态补偿不能假装没断过断线重连做到了只是“恢复收消息”还不够。比如任务从 10% 跑到 60% 的过程中网络断了恢复连接后前端只知道最后一条收到的消息是 10%而重新连接后 WebSocket 不会自动重放中间的消息。因此前端需要一个“状态补偿”机制。我推荐两种方案配合使用服务端在每次连接建立后立刻推送一个snapshot事件包含当前任务完整状态前端重连成功后如果几秒内没收到 snapshot就主动调用一次GET /api/tasks/{id}拉取全量状态。方案一体验最好因为首屏也能直接拿到最新状态不只是断线恢复。实现上服务端可以在 Redis 里存一个最近的状态快照连接时发送。前端收到 snapshot 时直接替换当前状态后续增量事件再覆盖逻辑是幂等的。这样用户断网几分钟回来页面也能立刻回到“正确的时间线”而不是永远停在一段旧数据上。4. 调试和线上排坑我在真实环境里遇到的事WebSocket 项目写起来“看起来很简单”但调起来比 HTTP 烦得多。HTTP 请求失败有状态码、有响应体WebSocket 断了常常只有一个莫名其妙的 close code甚至什么都没有。下面是我觉得最值得记录的几类问题。4.1 用测试客户端把消息流“看”出来浏览器 DevTools 的 Network 面板可以看到 WebSocket 帧但我不建议一开始就依赖它。更高效的方式是先用命令行客户端直接连后端把“后端能不能发消息”“消息内容是什么”先验证清楚再回前端调试。如果你装了wscatwscat -c ws://127.0.0.1:8000/ws/tasks/8f0c2f3e-.../?tokentest-token也可以用 Python 的websockets库写一个最小测试客户端import asyncio import json import websockets async def main(): uri ws://127.0.0.1:8000/ws/tasks/8f0c2f3e-.../?tokentest-token async with websockets.connect(uri) as ws: await ws.send(json.dumps({type: heartbeat})) async for raw in ws: print(raw) asyncio.run(main())这两招能把问题快速分成两层连得上但收不到消息问题多在 channel layer 或 group 名称不一致连不上问题多在路由、鉴权或反向代理。别一上来就怀疑前端代码。如果你用 Django Channels还可以用官方的测试客户端写自动化用例from channels.testing import WebSocketCommunicator from config.asgi import application async def test_task_push(): communicator WebSocketCommunicator( application, /ws/tasks/8f0c2f3e-.../?tokentest-token, ) connected, _ await communicator.connect() assert connected await communicator.send_json_to({type: heartbeat}) resp await communicator.receive_json_from() assert resp[type] heartbeat_ack await communicator.disconnect()WebSocket 测试客户端是排查消息字段、group 路由问题最有力的工具比自己拿 curl 盲猜靠谱多了。4.2 半开连接、心跳包与连接泄漏线上最容易遇到的坑是“连接泄漏”。现象很典型页面关了后端的连接数却一直在涨。这是因为很多关闭场景是“半开连接”——比如用户直接合上笔记本TCP 断开没有正常 FIN 包服务端永远收不到 close 事件。如果代码里只在disconnect中做 group_discard这些幽灵连接会一直留在频道组中。我对付这个问题的组合拳客户端断线重连做好指数退避服务端用 2.5 节的心跳机制超过 90 秒没收到任何消息就主动closedisconnect里一定要group_discard防止频道组越来越膨胀监控层面记录当前活跃连接数如果和在线用户数严重不匹配优先怀疑心跳没生效。另外要留意 Nginx 的proxy_read_timeout。如果设成默认的 60 秒哪怕后端和客户端都在正常传数据只要 60 秒内没有字节流动Nginx 就会主动掐断连接。前端表现为每 60 秒掉一次线。心跳包正好能解决这个问题因为每 30 秒就有字节流动反向代理的超时不会触发。4.3 反向 WebSocket当你的后端也要当客户端“反 WebSocket”在 Python 社区里通常指的是后端不再作为服务端等别人连而是作为客户端主动连别人的 WebSocket 接口。我实际遇到过一个场景上游有个文件变化通知服务需要保持一条 WebSocket 长连接接收事件再转成任务状态推给使用者。这里核心难点不是“连上”而是“断了怎么办”。我当时用websockets库写了一个后台常驻任务import asyncio import websockets async def upstream_listener(): retry 0 while True: try: async with websockets.connect(UPSTREAM_WS_URL) as ws: retry 0 async for raw in ws: await process_upstream_message(raw) except (websockets.ConnectionClosedError, OSError) as exc: retry 1 delay min(30, 2 ** retry) print(fupstream disconnected, retry in {delay}s: {exc}) await asyncio.sleep(delay)几个注意点必须在except里捕获 ConnectionClosedError否则进程会静默退出重连延迟指数退避避免上游抖动时全场疯狂重连如果上游消息密集消费速度跟不上应该把消息先放进asyncio.Queue由独立协程消费否则缓冲区积压会导致连接被系统关闭。这种“反向 WebSocket”场景其实很常见不只是文件变化消息推送、口令下发、远程控制指令都可能是上游主动推给后端。写过一次重连逻辑之后其他项目都能复用。4.4 反向代理和网关配置Nginx 升级头与超时WebSocket 走 Nginx 必须显式配置 Upgrade 头。省略这段前端会收到 400 或 502而且浏览器控制台里只有一个笼统的错误让人非常头疼。基础配置如下map $http_upgrade $connection_upgrade { default upgrade; close; } location /ws/ { proxy_pass http://django_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection $connection_upgrade; proxy_set_header Host $host; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }proxy_read_timeout和proxy_send_timeout我直接设成 1 小时。对任务状态推送这种长连接业务空闲不代表失效心跳包会在 30 秒内打断空闲状态所以不用担心超时导致误杀。如果你的内部网络还有负载均衡要确保所有后端实例能访问同一个 Redis。Django Channels 的 channel layer 会通过 Redis 把消息路由到持有对应连接的那个 worker所以不需要依赖 sticky session但前提是 Redis 必须是共享的。5. 一套可以直接抄的极简 Demo 与验收清单下面给出一套最小可运行的核心代码骨架。后端是 Django Channels前端是 React Hook都是生产环境验证过的写法。5.1 后端 Demo消费者与跨进程推送# consumers.py import json import time from channels.generic.websocket import AsyncWebsocketConsumer class TaskConsumer(AsyncWebsocketConsumer): async def connect(self): self.task_id self.scope[url_route][kwargs][task_id] self.group_name ftask_{self.task_id} self.last_seen time.time() await self.channel_layer.group_add(self.group_name, self.channel_name) await self.accept() await self.send_json({type: task.snapshot, payload: await load_task_snapshot(self.task_id)}) self.alive_check_task asyncio.create_task(self.check_alive_loop()) async def disconnect(self, code): self.alive_check_task.cancel() await self.channel_layer.group_discard(self.group_name, self.channel_name) async def receive(self, text_dataNone, bytes_dataNone): data json.loads(text_data) self.last_seen time.time() if data.get(type) heartbeat: await self.send_json({type: heartbeat_ack}) async def check_alive_loop(self): while True: await asyncio.sleep(30) if time.time() - self.last_seen 90: await self.close(code4400) break async def task_push(self, event): await self.send_json(event[event])# jobs.py from channels.layers import get_channel_layer from asgiref.sync import async_to_sync def send_task_event(task_id: str, event: dict): channel_layer get_channel_layer() async_to_sync(channel_layer.group_send)( ftask_{task_id}, {type: task.push, event: event}, )5.2 前端 Demo可用的 useWebSocket Hook参照 3.2 节的完整 Hook 即可这里补充一个关键使用姿势const { connected } useTaskWebSocket(taskId, { enabled: !!taskId, onEvent: (msg) { if (msg.type task.snapshot) { dispatch(loadTask(msg.payload)); } else if (msg.type task.progress) { dispatch(updateProgress(msg.payload.progress)); } }, });connected状态由onopen和onclose维护。页面顶部会显示“连接中”“已断开正在重连”“实时已连接”三种状态。这个小小的状态标识能省掉大量“为什么页面不动”的排查时间。5.3 验收清单按这个顺序逐项打勾场景预期结果验证方式新建任务页面立刻显示 pending不等轮询间隔打开浏览器 Network 面板看 WS 帧进度更新进度条平滑变化无卡顿后端每 1% 推一次观察页面渲染切到后台再回来连接不中断或能自动重连移动端切 App 后回到页面查看状态断网恢复显示重连中恢复后状态同步DevTools Network 切 Offline 再切回服务端重启客户端自动重连拿到新 snapshotkill worker 容器观察连接恢复多用户同时看一个任务各连接互不干扰消息不串号开两个浏览器 tab 同时查看长时间挂机连接不丢无幽灵连接观察 Nginx 连接数和后端日志最后再分享一个小技巧把 WebSocket 消息结构设计成和轮询接口返回的是同一种“事件格式”前端无论走 WebSocket 还是降级轮询都进入同一套onEvent处理逻辑。我在生产环境里保留了 2 秒间隔的轮询兜底当 WebSocket 连续失败超过 5 次就自动降级。这样即使某些网络环境会杀掉长连接用户也不会彻底看不到任务状态。等连接恢复再平滑切回 WebSocket。这种“实时优先、轮询兜底”的架构才是状态推送类功能真正上线后还能睡得着觉的形态。
