不用CeleryPRINTFILM進程內(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)典組合而是用進程內(nèi)任務(wù)平臺——調(diào)度器Scheduler、執(zhí)行器Executor、輪詢器Poller三件套 看門狗全部跑在 FastAPI 進程內(nèi)的 asyncio 事件循環(huán)里數(shù)據(jù)庫即隊列。本文詳解這套「輕量級任務(wù)平臺」的設(shè)計思路適合想理解異步任務(wù)系統(tǒng)架構(gòu)的新手閱讀。一、為什么不用 Celery短劇創(chuàng)作場景下任務(wù)的特點很鮮明任務(wù)數(shù)量不大同一用戶并發(fā)有限默認單用戶 4 個執(zhí)行槽全站默認 4 個 Worker 槽位任務(wù)耗時差異極大劇本摘要幾秒分鏡視頻可能幾分鐘到幾十分鐘需要強一致的狀態(tài)與計費每個任務(wù)都涉及余額預(yù)扣、按實結(jié)算、失敗退款。如果上 Celery就要多維護 Redis Broker、結(jié)果后端、獨立 Worker 進程還要處理序列化、重試、計費對賬跨進程同步的問題。PRINTFILM 選擇了更「單進程友好」的路線用 Postgres 的task_runs表當任務(wù)隊列用三個 asyncio 循環(huán)當 Worker用數(shù)據(jù)庫樂觀更新當鎖。所有任務(wù)記錄含步驟、事件、目標實體都由 models_tasks.py 中 4 張表承載TaskRun主記錄、TaskStep執(zhí)行步驟、TaskTarget關(guān)聯(lián)實體、TaskEvent只追加的生命周期事件日志。任務(wù)一進來就落庫進程重啟也不丟。二、任務(wù)狀態(tài)機一張表看懂流轉(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ù)若本進程沒有對應(yīng)協(xié)程就放回隊列recover_orphaned_tasks。這一步保證了進程崩潰后任務(wù)不會永遠卡死串行批次對賬短劇「尾幀銜接」模式下上一鏡完成才激活下一鏡見 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é)程——這就是進程內(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é)算多退少補異常則走_fail_task把錯誤寫回任務(wù)、關(guān)聯(lián)資產(chǎn)和分鏡同步觸發(fā)「后續(xù)串行分鏡全部作廢」。兩個新手容易忽略的巧思NIO 提交即走分鏡視頻這類長任務(wù)handler 只負責把請求提交給上游并把任務(wù)置為awaiting_poll然后立即釋放執(zhí)行槽——否則 4 個槽位會被幾分鐘的視頻生成占死中斷可重排捕獲CancelledError應(yīng)用重啟/熱更新時把任務(wù)改回pending重新排隊而不是丟單。五、輪詢器 PollerNIO 風格的 Selector輪詢循環(huán)在 poller.py作者直接把它類比成NIO 的 Selector不阻塞、不輪詢?nèi)咳蝿?wù)只處理「就緒集合」。每一輪默認 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é)算視頻成功后進入finalizing收尾窗口——下載成片、落 OSS、記錄真實 token 用量然后按實際用量結(jié)算。這個窗口特意設(shè)計成取消請求要等它完成executor.py 的 finalizing 窗口否則會出現(xiàn)「錢退了、片子也生成了」的兩失局面超時兜底輕量視頻任務(wù)輪詢超過 15 分鐘直接判失敗并退回預(yù)扣?！竿吮軙r間落庫」而非內(nèi)存計時是這套設(shè)計能水平擴展、重啟無損的關(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)整配置默認值含義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í)行器干活、輪詢器收活職責像微服務(wù)一樣清晰但共享同一個事件循環(huán)零 IPC 開銷 ?樂觀更新當分布式鎖一條帶rowcount判斷的 UPDATE 搞定搶占不需要 Redis Lock ?計費與任務(wù)生命周期深度綁定預(yù)扣 → 結(jié)算 → 失敗退款 → 收尾窗口防退款競態(tài)錢貨兩手都要硬。對于單/少副本、任務(wù)并發(fā)在個位數(shù)的業(yè)務(wù)AI 生成、短劇創(chuàng)作、內(nèi)容流水線這套「進程內(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），僅供參考