AsyncResult无法查实时进度，因state仅含PENDING/STARTED/SUCCESS等粗粒度状态；须任务内用Redis缓存主动上报进度，WebSocket后台协程轮询推送，前端防抖+超时关闭。
为什么不能直接用 AsyncResult 查 Celery 任务进度
因为 Celery 默认的 AsyncResult 只能拿到任务是否完成、结果或异常，state 字段只有 PENDING / STARTED / SUCCESS 等粗粒度状态，不包含「已处理 63%」这种实时进度。除非你手动改写后端存储逻辑，否则靠 task_id 调 result.state 是查不到百分比的。
真正可行的路径是：任务内部主动上报进度 → 存到共享存储（如 Redis）→ WebSocket 后端定时读取并推送给前端。
用 Redis + cache.set() 存进度，别碰数据库
每次任务执行中更新进度，必须选低延迟、高并发友好的存储。Django 的 cache（后端设为 Redis）最合适——比写数据库快两个数量级，且天然支持过期时间，避免陈旧进度堆积。
- 在异步任务函数里，用
cache.set(f"task_progress_{task_id}", {"percent": 45, "msg": "正在压缩第3个文件"}, timeout=300) - 前端通过 WebSocket 连接时，后端应立刻查一次该 key，把初始进度发过去，避免白屏等待
- 别用
cache.incr()直接累加百分比——它不保证原子性，多线程/多进程下容易错乱；老老实实用cache.set()覆盖整个进度 dict
Django Channels 的 WebsocketConsumer 怎么轮询进度
WebSocket 连接是长连接，但 Django Channels 不允许在 consumer 里开无限 while 循环阻塞事件循环。正确做法是用 asyncio.create_task() 启一个后台协程，配合 await asyncio.sleep() 控制轮询节奏。
立即学习“Python免费学习笔记（深入）”；
安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
class ProgressConsumer(WebsocketConsumer):
def connect(self):
self.accept()
self.task_id = self.scope["url_route"]["kwargs"]["task_id"]
# 启动后台轮询任务
asyncio.create_task(self._poll_progress())
<pre class="brush:php;toolbar:false;">async def _poll_progress(self):
while self.channel_layer: # 连接还活着
progress = cache.get(f"task_progress_{self.task_id}")
if progress:
await self.send(text_data=json.dumps(progress))
await asyncio.sleep(1.5) # 别太密，也别太疏
注意：self.channel_layer 是连接存活的简易判断依据；如果任务已结束，建议任务末尾主动删掉对应 cache key，避免前端收到重复 SUCCESS 消息。
前端 WebSocket 收到进度后怎么防抖和终止监听
前端容易犯两个错：一是没做消息去重，同一进度反复渲染；二是任务结束后没关连接，导致内存泄漏或持续收空数据。
- 用
Map缓存上一次收到的percent，只在变化时更新 DOM：if (data.percent !== lastPercent) { updateBar(data.percent); lastPercent = data.percent; } - 当收到
{"status": "SUCCESS"}或{"percent": 100}后，立刻调用ws.close()，并在onclose里清理定时器或监听器 - 别依赖后端“通知结束”——加个前端超时兜底：
setTimeout(() => ws.close(), 5 * 60 * 1000)，防止后端异常卡死
进度不是状态机，而是流式信号；后端不保证顺序，前端得自己容错。比如中间跳过 70% 直接到 95%，就得接受——这才是真实网络环境下的行为。
