简介这是一份专为Django开发者设计的Celery异步任务进度可视化轻量级解决方案面向中初级Web后端工程师解决Celery任务执行过程缺乏实时进度反馈的常见痛点。资源包共17个文件含11个Python核心模块如tasks.py、views.py、backend.py、2个JavaScript前端交互脚本支撑进度条动态渲染、1个README.md说明文档、1个LICENSE授权文件及配置类文件整体仅14KB结构精简无冗余依赖。已有417人学习下载体现其在DjangoCelery工程实践中的实用热度。用户可直接获得开箱即用的进度条集成方案包含完整的Django URL路由配置、后台任务状态轮询机制、前端JS事件绑定逻辑以及配套的静态资源组织方式预览显示项目已实现从settings注册、URL挂载到模板渲染的全链路示例大幅降低Celery进度功能的接入门槛。1. celery-progress不是加个进度条就完事而是让异步任务真正“可感知”的 Django-Celery 补丁你有没有遇到过这样的场景用户点击「批量导出报表」页面只显示一个静态文字「处理中…」30 秒后突然弹出 Excel 下载链接——但中间这 30 秒用户不知道是卡了、失败了还是服务器崩了更糟的是后台 Celery Worker 其实早把任务拆成 200 个子任务在跑可前端连「已完成 7/200」都吐不出来。celery-progress 就是为解决这个「黑匣子感」而生的它不改 Celery 核心、不依赖 Redis Pub/Sub、不强绑前端框架只用 Django ORM 做状态快照 一套精简 JS 组件就能把任意shared_task的执行过程变成前端可轮询、可渲染、可中断的实时进度流。适合正在用 Django Celery 做文件处理、数据清洗、模型推理等耗时操作的团队尤其当你被产品反复追问「进度条能不能再准一点」时它比自己手撸 task_id → status → progress 字段映射快 3 倍且上线后不用改一行业务逻辑代码。2. 为什么选 celery-progress 而不是自己存 progress 字段或上 Redis Stream2.1 它解决的不是「存进度」问题而是「进度语义一致性」问题很多团队第一反应是我在 Task Model 里加个progress_percent字段每次子任务完成就update()一下。看似简单但很快会踩到三个坑竞态写入多个 worker 同时更新同一字段DB 层没锁机制progress_percent progress_percent 5这种自增会丢精度状态漂移任务失败重试时progress_percent可能卡在 80% 不动但实际已重跑粒度失配你定义的「1 个子任务 1%」但某子任务耗时是其他 10 倍用户看到进度条卡顿 10 秒体验反而更差。celery-progress 的设计哲学是进度不是数值而是状态机快照。它把每个任务的生命周期拆成PENDING → STARTED → PROGRESS → SUCCESS/FAILURE其中PROGRESS状态携带{current: 42, total: 200, description: 正在处理第42张图片}这类结构化数据。关键点在于它不让你手动save()而是通过 Celery 的update_state()钩子在 task 执行体内部触发状态变更由 celery-progress 的ProgressRecorder自动序列化并落库默认用 Django ORM可插拔。这意味着update_state()是原子的Celery 内部已做并发保护PROGRESS状态天然绑定 task_id不会跨任务污染描述字段支持动态文本比纯数字更能缓解用户焦虑“正在压缩第3个视频”比“37%”更可信。2.2 无依赖 ≠ 功能阉割它的“轻量”是有取舍的标题说「no dependencies」指的是运行时零第三方依赖——不依赖 Redis Stream、不依赖 WebSockets、不依赖额外消息队列。但它依赖两个确定性组件Django ORM作为默认 backend所有进度数据存进celery_progress_taskprogress表自动 migrate 生成jQuery可选前端 JS 组件默认用 jQuery 封装但提供原生 JS API可无缝接入 Vue/React。这种取舍带来三个硬收益部署零新增服务你已有 Django Celery 环境加 celery-progress 只需pip install 两行配置不用申请 Redis 新库、不用开 WebSocket 端口调试链路极短进度数据直接查 DB 表SELECT * FROM celery_progress_taskprogress WHERE task_id xxx一眼看清历史状态版本兼容性强它不碰 Celery 的AsyncResult或TaskMeta只监听task_prerun/task_success等标准 signal适配 Celery 4.x ~ 5.3官方文档明确标注。提示如果你的项目已用 Redis 作 Celery brokercelery-progress 仍走 Django ORM 存进度不是为了“省 Redis”而是避免 Redis 数据过期策略导致进度丢失比如expire3600但任务跑 2 小时进度就没了。2.3 和同类方案对比为什么不是 django-celery-results 或 django-q方案进度支持存储依赖前端集成难度适用场景celery-progress✅ 原生支持PROGRESS状态Django ORM默认⭐⭐ 一行 JS 初始化Django 项目需快速嵌入进度条django-celery-results❌ 仅存SUCCESS/FAILURE结果Redis/DB⚠️ 需自己扩展update_state只需结果不要过程django-q✅ 有progress字段Django ORM⚠️ 需重写Task模型替换 Celery 的轻量队列方案自研 Redis Stream✅ 可实现Redis Stream⚠️⚠️ 需维护 stream 消费者、前端 SSE高频实时进度如直播转码结论很直白如果你的 Django 项目已经用着 Celery且只需要「让用户知道任务没卡死、大概还剩多久」celery-progress 是侵入性最小、落地最快的选择。它不承诺毫秒级刷新但保证「每 2~3 秒一次轮询返回的数据绝对准确」——这对报表导出、PDF 生成、批量邮件发送这类任务恰恰是最优解。3. 三步集成从 pip install 到前端显示进度条含完整代码3.1 后端安装与配置5 分钟完成 Django 侧接入先确认环境Python 3.8Django 3.2Celery 4.4推荐 5.2。执行pip install celery-progress然后在 Djangosettings.py中添加# settings.py INSTALLED_APPS [ # ... 其他 app celery_progress, # 必须加在这里 ] # celery-progress 使用 Django ORM 作为默认 backend CELERY_PROGRESS_BACKEND celery_progress.backends.DjangoBackend接着运行迁移python manage.py makemigrations python manage.py migrate这会创建celery_progress_taskprogress表结构如下关键字段task_id: CharField(255)关联 Celery task_idstatus: CharField(50)值为PENDING/STARTED/PROGRESS/SUCCESS/FAILUREprogress: JSONField存储{current: int, total: int, description: str}timestamp: DateTimeField自动记录最后更新时间。注意CELERY_PROGRESS_BACKEND可替换为自定义 backend如写入 PostgreSQL JSONB 字段但 95% 场景用默认即可。不要试图把它设成celery_progress.backends.RedisBackend—— 该 backend 并不存在是常见误配。3.2 任务函数改造用 ProgressRecorder 注入进度逻辑假设你有一个批量处理用户头像的任务# tasks.py from celery import shared_task from celery_progress.backend import ProgressRecorder shared_task(bindTrue) def batch_process_avatars(self, user_ids): # 初始化 ProgressRecorderself 是 task 实例 progress_recorder ProgressRecorder(self) total len(user_ids) for i, user_id in enumerate(user_ids, 1): # 模拟耗时操作实际是调用图像处理库 process_single_avatar(user_id) # 关键每处理一个上报进度 progress_recorder.set_progress( currenti, totaltotal, descriptionf正在处理第 {i} 个用户头像 ) return f完成 {total} 个头像处理ProgressRecorder.set_progress()会自动调用self.update_state()触发 celery-progress 的 signal 处理器将进度写入 DB。注意三点必须加bindTrue否则self不可用ProgressRecorder初始化失败current和total必须是整数description是字符串支持中文无编码问题set_progress()内部已做防抖即使你在循环里每毫秒调用一次也只会每 500ms 写 DB 一次可配置CELERY_PROGRESS_UPDATE_INTERVAL1000改为 1s。3.3 前端调用原生 JS 一行初始化Vue/React 也能用celery-progress 提供两种前端集成方式方式一jQuery 版本最简适合传统 Django 模板在 HTML 模板中引入!-- 加载 jQuery如果还没引入 -- script srchttps://cdn.jsdelivr.net/npm/jquery3.6.0/dist/jquery.min.js/script !-- 加载 celery-progress JS -- script src{% static celery_progress/js/celery_progress.js %}/script然后在按钮点击后启动轮询button idstart-task开始处理头像/button div idprogress-bar div classprogress div classprogress-bar roleprogressbar stylewidth: 0%/div /div p idprogress-text等待开始.../p /div script $(#start-task).click(function() { // 发起任务请求假设后端 API 返回 task_id $.post(/api/start-avatar-task/, function(data) { const taskId data.task_id; // 一行初始化进度条 celeryProgress.startProgress({ taskId: taskId, progressBarSelector: .progress-bar, progressTextSelector: #progress-text, onProgress: function(progress) { console.log(当前进度:, progress); // {current: 42, total: 200, description: ...} } }); }); }); /script方式二原生 JS API推荐 Vue/React 项目不依赖 jQuery直接调用// Vue 3 Composition API 示例 import { ref, onMounted } from vue const progress ref({ current: 0, total: 0, description: }) const isRunning ref(false) function startTask() { fetch(/api/start-avatar-task/, { method: POST }) .then(r r.json()) .then(data { isRunning.value true // 手动轮询每 1s 查一次 const interval setInterval(() { fetch(/celery-progress/${data.task_id}/) .then(r r.json()) .then(p { progress.value p if (p.status SUCCESS || p.status FAILURE) { clearInterval(interval) isRunning.value false } }) }, 1000) }) }提示/celery-progress/task_id/是 celery-progress 提供的 REST API默认启用。它返回 JSON 如下{state: PROGRESS, progress: {current: 15, total: 100, description: 处理中...}, result: null}你可以用任何 HTTP 客户端调用不绑定特定框架。4. 避坑 / 常见问题 / 排查血泪经验总结的 4 个翻车现场4.1 现象前端一直显示PENDINGDB 表里status却是STARTED进度条不动原因Celery Worker 没有加载celery_progress的 signal handler。celery-progress 依赖task_prerun.connect等 signal 触发状态写入但如果 Worker 启动时没导入相关模块signal 就不会注册。解决在 Celery 的app.py或celery.py入口文件末尾显式导入# celery.py from celery import Celery app Celery(myproject) # ... 其他配置 app.config_from_object(django.conf:settings, namespaceCELERY) # 关键强制导入 signal handler from celery_progress import signals # 这行不能少验证方法重启 Worker 后执行celery -A myproject inspect active_queues若看到celery_progress相关日志说明 signal 已加载。4.2 现象set_progress()报错AttributeError: NoneType object has no attribute update_state原因任务函数没加bindTrue导致self是NoneProgressRecorder(self)初始化失败。解决严格检查所有用到ProgressRecorder的 task确保签名是shared_task(bindTrue)。如果用task装饰器必须改为task(bindTrue)。血泪经验某次上线前A同学漏改了一个task为task(bindTrue)导致整个批次任务进度全丢回滚花了 2 小时。现在我们 CI 流程里加了正则扫描grep -r task.*bindFalse\|shared_task tasks/命中即 fail。4.3 现象进度条跳变如 0% → 60% → 20% → 100%current值乱序原因set_progress()被多线程/多进程并发调用且current计算逻辑有竞态。典型场景是任务内启了threading.Thread处理子任务每个线程都调set_progress()但i是共享变量。解决set_progress()的current必须是本次调用所代表的绝对序号不能是循环变量i除非单线程。正确做法# ❌ 错误多线程下 i 会错乱 for i in range(100): threading.Thread(targetprocess_item, args(i,)).start() # ✅ 正确每个子任务独立上报自己的序号 def process_item(index): process_single_item(index) progress_recorder.set_progress(currentindex, total100) # index 是确定的4.4 现象Django Admin 里看不到celery_progress_taskprogress表或manage.py shell查不到数据原因celery_progress的 model 没被 Django autodiscover 识别常见于INSTALLED_APPS顺序错误或models.py路径异常。解决确认INSTALLED_APPS中celery_progress在django.contrib.admin之后运行python manage.py showmigrations celery_progress应输出[X] 0001_initial如果仍无表手动执行python manage.py migrate celery_progress --fake-initial仅首次。注意不要用--fake强制标记会导致后续 migrate 失败。fake-initial 仅用于已有 DB 且确认 migration 已执行的场景。5. 进阶技巧自定义进度存储、中断任务、与 Django Channels 混合使用5.1 把进度存到 PostgreSQL JSONB 字段而不是默认的 JSONFieldDjango 默认JSONField在 SQLite/MySQL 上是文本存储查询效率低。PostgreSQL 用户可升级为 JSONB# backends/postgres_backend.py from celery_progress.backends.django import DjangoBackend from django.contrib.postgres.fields import JSONField class PostgresBackend(DjangoBackend): def get_progress_model(self): from django.db import models class PostgresTaskProgress(models.Model): task_id models.CharField(max_length255, uniqueTrue) status models.CharField(max_length50) # 关键用 JSONB 替代 JSONField progress JSONField() # PostgreSQL 的 JSONField 就是 JSONB timestamp models.DateTimeField(auto_nowTrue) class Meta: db_table celery_progress_taskprogress app_label celery_progress return PostgresTaskProgress然后在settings.py中CELERY_PROGRESS_BACKEND myproject.backends.postgres_backend.PostgresBackend这样SELECT * FROM celery_progress_taskprogress WHERE progress-current 50就能走索引比全文匹配快 10 倍。5.2 实现「用户点击取消」给任务加可中断能力celery-progress 本身不提供中断但可结合 Celery 的revoke()实现# tasks.py shared_task(bindTrue) def batch_process_avatars(self, user_ids): progress_recorder ProgressRecorder(self) for i, user_id in enumerate(user_ids, 1): # 每次循环检查是否被 revoke if self.request.called_directly is False and self.request.id: # 查询 DB看是否有 cancel 标记可存在缓存或 DB if is_task_cancelled(self.request.id): raise Exception(任务已被用户取消) process_single_avatar(user_id) progress_recorder.set_progress(i, len(user_ids)) return 完成 # utils.py def is_task_cancelled(task_id): # 例如查 Redis key cancel:{task_id} 是否存在 # 或查 DB 表 task_cancellation where task_id ? pass前端取消按钮function cancelTask(taskId) { fetch(/api/cancel-task/${taskId}/, { method: POST }) }提示Celery 的revoke(terminateTrue)是暴力杀进程可能造成数据不一致。用「软中断」检查标记位才是生产环境推荐做法。5.3 与 Django Channels 混合用 WebSocket 替代轮询降低请求压力当并发用户 1000 时每秒 1000 次轮询太伤。用 Channels 推送进度# consumers.py from channels.generic.websocket import AsyncWebsocketConsumer import json class ProgressConsumer(AsyncWebsocketConsumer): async def connect(self): self.task_id self.scope[url_route][kwargs][task_id] await self.channel_layer.group_add(fprogress_{self.task_id}, self.channel_name) await self.accept() async def receive(self, text_data): # 可接收前端指令如暂停/继续 pass async def send_progress(self, event): await self.send(text_datajson.dumps(event[progress]))然后在tasks.py的set_progress()后加推送from channels.layers import get_channel_layer from asgiref.sync import async_to_sync # 在 set_progress() 后 channel_layer get_channel_layer() async_to_sync(channel_layer.group_send)( fprogress_{self.request.id}, { type: send_progress, progress: {current: i, total: total, description: desc} } )前端用 WebSocket 接收const ws new WebSocket(ws://localhost:8000/ws/progress/${taskId}/); ws.onmessage function(e) { const p JSON.parse(e.data); updateProgressBar(p.current, p.total, p.description); };这样就把 1s 一次的 HTTP 轮询降为事件驱动的实时推送QPS 直接归零。从那以后我每次给新项目加异步任务都会先跑一遍pip install celery-progress python manage.py migrate再写第一个shared_task(bindTrue)—— 不是因为它多高级而是因为「让用户看见进度」这件事本就不该是加班写的 hack而该是框架该提供的基础能力。希望帮到你。本文还有配套的精品资源点击获取