不用CeleryPRINTFILM進(jìn)程內(nèi)任務(wù)平臺調(diào)度/執(zhí)行/輪詢?nèi)自O(shè)計詳解【免費下載鏈接】printfilmPRINTFILMAI 視頻獲客與 AI短劇創(chuàng)作平臺項目地址: https://gitcode.com/gh_mirrors/pr/printfilmPRINTFILM是一個 AI 視頻獲客與 AI 短劇創(chuàng)作平臺從劇本、分鏡規(guī)劃到 AI 生圖、分鏡視頻生成全部由平臺后臺任務(wù)驅(qū)動。它沒有使用 Celery Redis 這套經(jīng)典組合而是用進(jìn)程內(nèi)任務(wù)平臺——調(diào)度器Scheduler、執(zhí)行器Executor、輪詢器Poller三件套 看門狗全部跑在 FastAPI 進(jìn)程內(nèi)的 asyncio 事件循環(huán)里數(shù)據(jù)庫即隊列。本文詳解這套「輕量級任務(wù)平臺」的設(shè)計思路適合想理解異步任務(wù)系統(tǒng)架構(gòu)的新手閱讀。一、為什么不用 Celery短劇創(chuàng)作場景下任務(wù)的特點很鮮明任務(wù)數(shù)量不大同一用戶并發(fā)有限默認(rèn)單用戶 4 個執(zhí)行槽全站默認(rèn) 4 個 Worker 槽位任務(wù)耗時差異極大劇本摘要幾秒分鏡視頻可能幾分鐘到幾十分鐘需要強(qiáng)一致的狀態(tài)與計費每個任務(wù)都涉及余額預(yù)扣、按實結(jié)算、失敗退款。如果上 Celery就要多維護(hù) Redis Broker、結(jié)果后端、獨立 Worker 進(jìn)程還要處理序列化、重試、計費對賬跨進(jìn)程同步的問題。PRINTFILM 選擇了更「單進(jìn)程友好」的路線用 Postgres 的task_runs表當(dāng)任務(wù)隊列用三個 asyncio 循環(huán)當(dāng) Worker用數(shù)據(jù)庫樂觀更新當(dāng)鎖。所有任務(wù)記錄含步驟、事件、目標(biāo)實體都由 models_tasks.py 中 4 張表承載TaskRun主記錄、TaskStep執(zhí)行步驟、TaskTarget關(guān)聯(lián)實體、TaskEvent只追加的生命周期事件日志。任務(wù)一進(jìn)來就落庫進(jìn)程重啟也不丟。二、任務(wù)狀態(tài)機(jī)一張表看懂流轉(zhuǎn)整個平臺圍繞TaskRun.status字段做狀態(tài)流轉(zhuǎn)核心狀態(tài)只有 7 個pending → leased → running → awaiting_poll → succeeded / failed └─────────────── cancel_requested任意時刻可請求取消pending已入隊等待調(diào)度leased調(diào)度器已「租約」搶占寫lease_token 10 分鐘lease_untilrunning執(zhí)行器已開跑先完成計費預(yù)扣awaiting_poll已提交上游如視頻生成 API轉(zhuǎn)入輪詢等待不占執(zhí)行槽位succeeded / failed / cancelled三個終態(tài)終態(tài)一律觸發(fā)計費結(jié)算。每次狀態(tài)變化都會往TaskEvent追加一條事件task.created、task.leased、task.started、task.failed、task.recovered……這就是用戶在任務(wù)詳情頁看到的「任務(wù)時間線」。三、調(diào)度器 Scheduler租約搶占 雙級并發(fā)閘調(diào)度循環(huán)在 scheduler.py 中每 1 秒 tick 一次每輪做 5 件事孤兒回收每 30 秒掃描「租約過期」或「running 超過 30 秒無心跳」的任務(wù)若本進(jìn)程沒有對應(yīng)協(xié)程就放回隊列recover_orphaned_tasks。這一步保證了進(jìn)程崩潰后任務(wù)不會永遠(yuǎn)卡死串行批次對賬短劇「尾幀銜接」模式下上一鏡完成才激活下一鏡見 service.py失效任務(wù)清掃分鏡被刪/已重新生成的在途任務(wù)自動作廢并退款處理取消請求把cancel_requested的本地協(xié)程 cancel 掉搶占到期任務(wù)查出next_action_at now的候選按「優(yōu)先級 → 創(chuàng)建時間」排序然后做樂觀更新租約——UPDATE ... WHERE id? AND status?rowcount ! 1就放棄。這種寫法在多副本部署下也能天然防重復(fù)執(zhí)行。搶占前有兩道并發(fā)閘閘配置作用全站 Worker 槽位task_runtime_max_concurrency 4限制同時 running 的任務(wù)總數(shù)單用戶并發(fā)task_user_max_concurrency 4防止單個用戶刷滿全站搶到任務(wù)后只是asyncio.create_task(execute_task_run(task_id))執(zhí)行體本身是普通協(xié)程——這就是進(jìn)程內(nèi) Worker的全部秘密。四、執(zhí)行器 Executor計費預(yù)扣 Handler 注冊表執(zhí)行入口是 executor.py 的execute_task_run流程像一個「三明治」上層開跑前查處理器是否存在 →freeze_for_task按預(yù)估凍結(jié)余額。余額不足直接以insufficient_balance失敗前端可引導(dǎo)充值中層開跑按domain task_type從 handlers.py 的注冊表找到協(xié)程處理器。平臺把 20 多種任務(wù)類型都掛在這里——漫劇劇本摘要、分鏡規(guī)劃、分鏡視頻、資產(chǎn)生圖/生視頻、科普流水線、甚至mock_delay測試任務(wù)下層收斂handler 返回后按結(jié)果收斂到終態(tài)并settle_task結(jié)算多退少補(bǔ)異常則走_(dá)fail_task把錯誤寫回任務(wù)、關(guān)聯(lián)資產(chǎn)和分鏡同步觸發(fā)「后續(xù)串行分鏡全部作廢」。兩個新手容易忽略的巧思NIO 提交即走分鏡視頻這類長任務(wù)handler 只負(fù)責(zé)把請求提交給上游并把任務(wù)置為awaiting_poll然后立即釋放執(zhí)行槽——否則 4 個槽位會被幾分鐘的視頻生成占死中斷可重排捕獲CancelledError應(yīng)用重啟/熱更新時把任務(wù)改回pending重新排隊而不是丟單。五、輪詢器 PollerNIO 風(fēng)格的 Selector輪詢循環(huán)在 poller.py作者直接把它類比成NIO 的 Selector不阻塞、不輪詢?nèi)咳蝿?wù)只處理「就緒集合」。每一輪默認(rèn) 8 秒select 到期 channel只查status awaiting_poll且next_action_at now的任務(wù)批量上限task_poll_max_concurrency 20并發(fā)非阻塞查詢用信號量 asyncio.gather并發(fā)查上游狀態(tài)。上游仍在running/queued就寫回退避時間next_action_at推遲 8 秒下輪再來終態(tài)結(jié)算視頻成功后進(jìn)入finalizing收尾窗口——下載成片、落 OSS、記錄真實 token 用量然后按實際用量結(jié)算。這個窗口特意設(shè)計成取消請求要等它完成executor.py 的 finalizing 窗口否則會出現(xiàn)「錢退了、片子也生成了」的兩失局面超時兜底輕量視頻任務(wù)輪詢超過 15 分鐘直接判失敗并退回預(yù)扣?！竿吮軙r間落庫」而非內(nèi)存計時是這套設(shè)計能水平擴(kuò)展、重啟無損的關(guān)鍵。六、看門狗 Watchdog給循環(huán)上保險再穩(wěn)的循環(huán)也可能卡死。runtime.py 里的看門狗每 5 秒檢查一次調(diào)度器或輪詢器的心跳上次 tick 距今的秒數(shù)超過閾值60s / 600s或者循環(huán)協(xié)程已退出就軟重啟對應(yīng)循環(huán)保留正在跑的任務(wù)。三個循環(huán)外加應(yīng)用啟停鉤子構(gòu)成完整生命周期啟動main.py 的start_task_runtime()依次拉起調(diào)度器、輪詢器、看門狗停止優(yōu)雅關(guān)閉時取消所有執(zhí)行協(xié)程任務(wù)自動重排。管理后臺還有一個運行時摘要接口把「槽位占用 三循環(huán)存活 心跳年齡」暴露成健康狀態(tài)方便運維一眼判斷任務(wù)平臺是否卡死。七、關(guān)鍵配置一覽所有旋鈕都集中在 config.py 且可在管理后臺動態(tài)調(diào)整配置默認(rèn)值含義task_runtime_max_concurrency4全站同時執(zhí)行的任務(wù)數(shù)task_user_max_concurrency4單用戶同時執(zhí)行上限task_poll_max_concurrency20輪詢器每輪并發(fā)查詢數(shù)task_runtime_orphan_check_sec30孤兒掃描間隔task_runtime_tick_stale_sec60調(diào)度心跳過期閾值task_runtime_watchdog_interval_sec5看門狗檢查間隔八、小結(jié)這套設(shè)計能學(xué)到什么?數(shù)據(jù)庫即隊列任務(wù)狀態(tài)、租約、退避時間全部落庫天然支持重啟恢復(fù)與多副本 ?三個循環(huán)各司其職調(diào)度器搶活、執(zhí)行器干活、輪詢器收活職責(zé)像微服務(wù)一樣清晰但共享同一個事件循環(huán)零 IPC 開銷 ?樂觀更新當(dāng)分布式鎖一條帶rowcount判斷的 UPDATE 搞定搶占不需要 Redis Lock ?計費與任務(wù)生命周期深度綁定預(yù)扣 → 結(jié)算 → 失敗退款 → 收尾窗口防退款競態(tài)錢貨兩手都要硬。對于單/少副本、任務(wù)并發(fā)在個位數(shù)的業(yè)務(wù)AI 生成、短劇創(chuàng)作、內(nèi)容流水線這套「進(jìn)程內(nèi)三件套」比 Celery 更輕、更好調(diào)試、更省錢。想動手研究建議按 backend/app/services/tasks/ 目錄從service.py建任務(wù)→scheduler.py搶任務(wù)→executor.py跑任務(wù)→poller.py收任務(wù)的順序讀配合 backend/tests/ 里的test_task_poller_dispatch.py、test_task_runtime_watchdog.py等測試用例很快就能摸清全貌?！久赓M下載鏈接】printfilmPRINTFILMAI 視頻獲客與 AI短劇創(chuàng)作平臺項目地址: https://gitcode.com/gh_mirrors/pr/printfilm創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考