不用CeleryPRINTFILM进程内任务平台调度/执行/轮询三件套设计详解【免费下载链接】printfilmPRINTFILMAI 视频获客与 AI短剧创作平台项目地址: https://gitcode.com/gh_mirrors/pr/printfilmPRINTFILM是一个 AI 视频获客与 AI 短剧创作平台从剧本、分镜规划到 AI 生图、分镜视频生成全部由平台后台任务驱动。它没有使用 Celery Redis 这套经典组合而是用进程内任务平台——调度器Scheduler、执行器Executor、轮询器Poller三件套 看门狗全部跑在 FastAPI 进程内的 asyncio 事件循环里数据库即队列。本文详解这套「轻量级任务平台」的设计思路适合想理解异步任务系统架构的新手阅读。一、为什么不用 Celery短剧创作场景下任务的特点很鲜明任务数量不大同一用户并发有限默认单用户 4 个执行槽全站默认 4 个 Worker 槽位任务耗时差异极大剧本摘要几秒分镜视频可能几分钟到几十分钟需要强一致的状态与计费每个任务都涉及余额预扣、按实结算、失败退款。如果上 Celery就要多维护 Redis Broker、结果后端、独立 Worker 进程还要处理序列化、重试、计费对账跨进程同步的问题。PRINTFILM 选择了更「单进程友好」的路线用 Postgres 的task_runs表当任务队列用三个 asyncio 循环当 Worker用数据库乐观更新当锁。所有任务记录含步骤、事件、目标实体都由 models_tasks.py 中 4 张表承载TaskRun主记录、TaskStep执行步骤、TaskTarget关联实体、TaskEvent只追加的生命周期事件日志。任务一进来就落库进程重启也不丢。二、任务状态机一张表看懂流转整个平台围绕TaskRun.status字段做状态流转核心状态只有 7 个pending → leased → running → awaiting_poll → succeeded / failed └─────────────── cancel_requested任意时刻可请求取消pending已入队等待调度leased调度器已「租约」抢占写lease_token 10 分钟lease_untilrunning执行器已开跑先完成计费预扣awaiting_poll已提交上游如视频生成 API转入轮询等待不占执行槽位succeeded / failed / cancelled三个终态终态一律触发计费结算。每次状态变化都会往TaskEvent追加一条事件task.created、task.leased、task.started、task.failed、task.recovered……这就是用户在任务详情页看到的「任务时间线」。三、调度器 Scheduler租约抢占 双级并发闸调度循环在 scheduler.py 中每 1 秒 tick 一次每轮做 5 件事孤儿回收每 30 秒扫描「租约过期」或「running 超过 30 秒无心跳」的任务若本进程没有对应协程就放回队列recover_orphaned_tasks。这一步保证了进程崩溃后任务不会永远卡死串行批次对账短剧「尾帧衔接」模式下上一镜完成才激活下一镜见 service.py失效任务清扫分镜被删/已重新生成的在途任务自动作废并退款处理取消请求把cancel_requested的本地协程 cancel 掉抢占到期任务查出next_action_at now的候选按「优先级 → 创建时间」排序然后做乐观更新租约——UPDATE ... WHERE id? AND status?rowcount ! 1就放弃。这种写法在多副本部署下也能天然防重复执行。抢占前有两道并发闸闸配置作用全站 Worker 槽位task_runtime_max_concurrency 4限制同时 running 的任务总数单用户并发task_user_max_concurrency 4防止单个用户刷满全站抢到任务后只是asyncio.create_task(execute_task_run(task_id))执行体本身是普通协程——这就是进程内 Worker的全部秘密。四、执行器 Executor计费预扣 Handler 注册表执行入口是 executor.py 的execute_task_run流程像一个「三明治」上层开跑前查处理器是否存在 →freeze_for_task按预估冻结余额。余额不足直接以insufficient_balance失败前端可引导充值中层开跑按domain task_type从 handlers.py 的注册表找到协程处理器。平台把 20 多种任务类型都挂在这里——漫剧剧本摘要、分镜规划、分镜视频、资产生图/生视频、科普流水线、甚至mock_delay测试任务下层收敛handler 返回后按结果收敛到终态并settle_task结算多退少补异常则走_fail_task把错误写回任务、关联资产和分镜同步触发「后续串行分镜全部作废」。两个新手容易忽略的巧思NIO 提交即走分镜视频这类长任务handler 只负责把请求提交给上游并把任务置为awaiting_poll然后立即释放执行槽——否则 4 个槽位会被几分钟的视频生成占死中断可重排捕获CancelledError应用重启/热更新时把任务改回pending重新排队而不是丢单。五、轮询器 PollerNIO 风格的 Selector轮询循环在 poller.py作者直接把它类比成NIO 的 Selector不阻塞、不轮询全部任务只处理「就绪集合」。每一轮默认 8 秒select 到期 channel只查status awaiting_poll且next_action_at now的任务批量上限task_poll_max_concurrency 20并发非阻塞查询用信号量 asyncio.gather并发查上游状态。上游仍在running/queued就写回退避时间next_action_at推迟 8 秒下轮再来终态结算视频成功后进入finalizing收尾窗口——下载成片、落 OSS、记录真实 token 用量然后按实际用量结算。这个窗口特意设计成取消请求要等它完成executor.py 的 finalizing 窗口否则会出现「钱退了、片子也生成了」的两失局面超时兜底轻量视频任务轮询超过 15 分钟直接判失败并退回预扣。「退避时间落库」而非内存计时是这套设计能水平扩展、重启无损的关键。六、看门狗 Watchdog给循环上保险再稳的循环也可能卡死。runtime.py 里的看门狗每 5 秒检查一次调度器或轮询器的心跳上次 tick 距今的秒数超过阈值60s / 600s或者循环协程已退出就软重启对应循环保留正在跑的任务。三个循环外加应用启停钩子构成完整生命周期启动main.py 的start_task_runtime()依次拉起调度器、轮询器、看门狗停止优雅关闭时取消所有执行协程任务自动重排。管理后台还有一个运行时摘要接口把「槽位占用 三循环存活 心跳年龄」暴露成健康状态方便运维一眼判断任务平台是否卡死。七、关键配置一览所有旋钮都集中在 config.py 且可在管理后台动态调整配置默认值含义task_runtime_max_concurrency4全站同时执行的任务数task_user_max_concurrency4单用户同时执行上限task_poll_max_concurrency20轮询器每轮并发查询数task_runtime_orphan_check_sec30孤儿扫描间隔task_runtime_tick_stale_sec60调度心跳过期阈值task_runtime_watchdog_interval_sec5看门狗检查间隔八、小结这套设计能学到什么✅数据库即队列任务状态、租约、退避时间全部落库天然支持重启恢复与多副本 ✅三个循环各司其职调度器抢活、执行器干活、轮询器收活职责像微服务一样清晰但共享同一个事件循环零 IPC 开销 ✅乐观更新当分布式锁一条带rowcount判断的 UPDATE 搞定抢占不需要 Redis Lock ✅计费与任务生命周期深度绑定预扣 → 结算 → 失败退款 → 收尾窗口防退款竞态钱货两手都要硬。对于单/少副本、任务并发在个位数的业务AI 生成、短剧创作、内容流水线这套「进程内三件套」比 Celery 更轻、更好调试、更省钱。想动手研究建议按 backend/app/services/tasks/ 目录从service.py建任务→scheduler.py抢任务→executor.py跑任务→poller.py收任务的顺序读配合 backend/tests/ 里的test_task_poller_dispatch.py、test_task_runtime_watchdog.py等测试用例很快就能摸清全貌。【免费下载链接】printfilmPRINTFILMAI 视频获客与 AI短剧创作平台项目地址: https://gitcode.com/gh_mirrors/pr/printfilm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考