ArchiveBox 耐用任務(wù)隊(duì)列模型 ModelWithQueue 源碼解析status、retry_at 與原子租約機(jī)制【免費(fèi)下載鏈接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...項(xiàng)目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox本文以 ArchiveBox 的archivebox.workers.models模塊API 文檔見(jiàn) archivebox.workers.models.md實(shí)現(xiàn)見(jiàn) workers/models.py為核心深入講解 ArchiveBox 調(diào)度系統(tǒng)賴(lài)以運(yùn)轉(zhuǎn)的耐用隊(duì)列durable queue協(xié)議status/retry_at雙字段狀態(tài)模型、pause/resume生命周期控制、基于條件更新的原子租約lease與安全更新機(jī)制以及它們?cè)?Crawl、Snapshot、Binary 等具體模型和 runner 運(yùn)行器中的實(shí)際調(diào)用方式。讀完本文你將掌握 ArchiveBox 分布式/多進(jìn)程調(diào)度中誰(shuí)擁有任務(wù)、任務(wù)何時(shí)重試、如何防止重復(fù)執(zhí)行的底層實(shí)現(xiàn)原理并了解如何在自己的 Django 模型上復(fù)用這套隊(duì)列協(xié)議。模塊定位調(diào)度系統(tǒng)的公共隊(duì)列協(xié)議ArchiveBox 使用 Django ORM 作為任務(wù)調(diào)度的中心化狀態(tài)存儲(chǔ)爬蟲(chóng)任務(wù)Crawl、頁(yè)面快照Snapshot、依賴(lài)二進(jìn)制安裝Binary都圍繞隊(duì)列行運(yùn)轉(zhuǎn)由 runner.py 中的運(yùn)行器循環(huán)消費(fèi)。archivebox.workers.models就是這套體系的最小公共部分——它只定義了一個(gè)抽象基類(lèi)ModelWithQueue其模塊 docstring 明確寫(xiě)道Durable queue fields and atomic lease operations shared by work rows. Concrete models own lifecycle transitions. This mixin only owns the common database queue protocol: status, retry_at, pause/resume, and claims.也就是說(shuō)具體模型自己決定生命周期狀態(tài)遷移規(guī)則ModelWithQueue只負(fù)責(zé)提供通用的數(shù)據(jù)庫(kù)隊(duì)列協(xié)議——狀態(tài)字段、重試時(shí)間、暫停/恢復(fù)、以及認(rèn)領(lǐng)claim操作。這種mixin 管協(xié)議、具體模型管遷移的職責(zé)劃分是理解整個(gè) workers 模塊的關(guān)鍵。隊(duì)列狀態(tài)協(xié)議DefaultStatusChoices模塊首先定義了默認(rèn)的狀態(tài)枚舉class DefaultStatusChoices(models.TextChoices): QUEUED queued, Queued STARTED started, Started PAUSED paused, Paused SEALED sealed, Sealed四個(gè)狀態(tài)的含義如下?tīng)顟B(tài)枚舉值含義QUEUEDqueued任務(wù)已入隊(duì)等待 worker 認(rèn)領(lǐng)STARTEDstarted任務(wù)正在處理中活躍狀態(tài)PAUSEDpaused任務(wù)被用戶暫停不再調(diào)度SEALEDsealed任務(wù)已封存終態(tài)不再處理DefaultStatusChoices繼承自 Django 的models.TextChoices因此同時(shí)具備枚舉常量DefaultStatusChoices.QUEUED與字段 choices.choices兩套能力。需要說(shuō)明的是這只是默認(rèn)協(xié)議——使用方可以通過(guò)自定義StatusChoices完全重定義狀態(tài)集見(jiàn)下文 Binary 模型的定制因此這四態(tài)是調(diào)度系統(tǒng)的常規(guī)約定而非鐵律。兩個(gè)核心隊(duì)列字段與默認(rèn)值工廠ModelWithQueue只有兩個(gè)自有數(shù)據(jù)庫(kù)字段它們構(gòu)成了隊(duì)列的雙支柱default_status_field: models.CharField models.CharField( choicesDefaultStatusChoices.choices, max_length15, defaultDefaultStatusChoices.QUEUED, nullFalse, blankFalse, db_indexTrue, ) default_retry_at_field: models.DateTimeField models.DateTimeField(defaulttimezone.now, nullTrue, blankTrue, db_indexTrue)statusCharFieldmax_length15任務(wù)的當(dāng)前狀態(tài)默認(rèn)QUEUEDnullFalse, blankFalse強(qiáng)制非空且db_indexTrue保證按狀態(tài)過(guò)濾的高效性。retry_atDateTimeField可空默認(rèn)當(dāng)前時(shí)間任務(wù)何時(shí)到期可被認(rèn)領(lǐng)。它同時(shí)承擔(dān)了到期時(shí)間與租約鎖雙重職責(zé)下文詳述同樣帶數(shù)據(jù)庫(kù)索引。字段本身通過(guò)deconstruct()參數(shù)以模塊級(jí)默認(rèn)值工廠的形式定義default_status_field/default_retry_at_field方便各模型復(fù)用。ModelWithQueue內(nèi)部據(jù)此聲明字段status: models.CharField models.CharField(**default_status_field.deconstruct()[3]) retry_at: models.DateTimeField models.DateTimeField(**default_retry_at_field.deconstruct()[3])使用方可以在此基礎(chǔ)上覆蓋參數(shù)例如 Binary 模型調(diào)用ModelWithQueue.StatusField(choicesStatusChoices.choices, defaultStatusChoices.QUEUED, max_length16)見(jiàn) machine/models.py微調(diào)max_length。兩個(gè)關(guān)鍵常量RETRY_AT_MAX datetime(9999, 1, 1, tzinfoUTC) ACTIVE_STATE_LEASE_SECONDS 60RETRY_AT_MAX一個(gè)遠(yuǎn)未來(lái)時(shí)間戳語(yǔ)義上等價(jià)于永不重試。pause()會(huì)把retry_at置為該值從而讓被暫停的任務(wù)在get_queue()查詢(xún)中永遠(yuǎn)不滿足retry_at__ltenow條件實(shí)現(xiàn)不可見(jiàn)即不調(diào)度。ACTIVE_STATE_LEASE_SECONDS 60活躍狀態(tài)的默認(rèn)租約時(shí)長(zhǎng)。當(dāng) Crawl 保持STARTED狀態(tài)繼續(xù)推進(jìn)子任務(wù)時(shí)runner 會(huì)用它刷新父行的租約見(jiàn) runner.py避免父行租約過(guò)期被其他 worker 搶走。模塊還導(dǎo)出了logger模塊日志器、MODULE_PATH/REPO_ROOT/PACKAGE_ROOT用于save()覆蓋中定位調(diào)用方棧幀見(jiàn)下文這些是支撐調(diào)試告警的輔助常量。ModelWithQueue抽象基類(lèi)的完整 APIModelWithQueue繼承django.db.models.Model其Meta聲明app_label workers且abstract True——它不產(chǎn)生數(shù)據(jù)表只為子類(lèi)提供字段與方法從倉(cāng)庫(kù)看 workers/migrations 目錄僅有__init__.py正因抽象模型不生成遷移。下面按職責(zé)分組介紹其完整 API。類(lèi)級(jí)狀態(tài)常量類(lèi)屬性默認(rèn)值說(shuō)明StatusChoicesDefaultStatusChoices狀態(tài)枚舉子類(lèi)可覆蓋INITIAL_STATEQUEUED初始狀態(tài)ACTIVE_STATESTARTED活躍狀態(tài)FINAL_STATES(SEALED,)終態(tài)集合FINAL_OR_ACTIVE_STATES(*FINAL_STATES, ACTIVE_STATE)終態(tài) 活躍態(tài)warn_on_save_outside_runnerTrue是否在 runner 進(jìn)程外save()時(shí)告警實(shí)例屬性RETRY_AT與STATE只是retry_at、status的屬性別名提供統(tǒng)一的讀寫(xiě)接口。生命周期操作pause / resume / bump_retry_atdef pause(self, *, save: bool True) - bool: paused_state getattr(self.StatusChoices, PAUSED, None) if paused_state is None or self.status in self.FINAL_STATES or self.is_paused: return False previous_status self.status self.status paused_state self.retry_at RETRY_AT_MAX if save: return self.safe_update( {status: paused_state, retry_at: RETRY_AT_MAX}, extra_filter{status: previous_status}, ) return Truepause()將狀態(tài)置為PAUSED并把retry_at推到RETRY_AT_MAX。三個(gè)保護(hù)條件——枚舉未定義PAUSED、已是終態(tài)、已處于暫?！紩?huì)直接返回False。寫(xiě)庫(kù)時(shí)使用safe_update且extra_filter{status: previous_status}確保只有仍處于原狀態(tài)的行才被更新并發(fā)安全。resume(whenNone)僅當(dāng)當(dāng)前is_paused時(shí)才生效把狀態(tài)改回QUEUEDretry_at設(shè)為when默認(rèn)timezone.now()即立即恢復(fù)調(diào)度同樣帶extra_filter{status: paused_state}的守衛(wèi)。調(diào)用方可以指定when實(shí)現(xiàn)定時(shí)恢復(fù)。bump_retry_at(seconds10)把retry_at從當(dāng)前時(shí)刻順延 N 秒用于簡(jiǎn)單的指數(shù)退避或失敗重試。is_paused判斷當(dāng)前狀態(tài)是否為枚舉中定義的PAUSED。隊(duì)列查詢(xún)與原子租約get_queue / claim_for_worker / claim_processing_lock這是模塊最核心的并發(fā)機(jī)制全部圍繞retry_at實(shí)現(xiàn)誰(shuí)先到期誰(shuí)被處理、誰(shuí)先更新誰(shuí)擁有classmethod def get_queue(cls): return cls.objects.filter(retry_at__ltetimezone.now()).order_by(retry_at) classmethod def claim_for_worker(cls, obj: ModelWithQueue, lock_seconds: int 60) - bool: now timezone.now() lock_until now timedelta(secondslock_seconds) updated cls.objects.filter(pkobj.pk, retry_atobj.retry_at, retry_at__ltenow).update( retry_atlock_until, modified_atnow, ) if updated 1: obj.retry_at lock_until cast(Any, obj).modified_at now return updated 1get_queue()取出所有到期retry_at now的行并按retry_at升序排列構(gòu)成待處理隊(duì)列。這個(gè)查詢(xún)天然排除了PAUSEDretry_at為遠(yuǎn)未來(lái)和未來(lái)才到期的任務(wù)。claim_for_worker()認(rèn)領(lǐng)的核心是一個(gè)**樂(lè)觀鎖 CAS比較并交換**操作。更新條件同時(shí)包含pk、retry_atobj.retry_at讀到的舊值和retry_at__ltenow未過(guò)期。只有恰好更新 1 行updated 1才算認(rèn)領(lǐng)成功并把retry_at推進(jìn)到當(dāng)前時(shí)間 lock_seconds相當(dāng)于租約到期時(shí)間。兩個(gè)并發(fā) worker 同時(shí)認(rèn)領(lǐng)同一行時(shí)后執(zhí)行的 UPDATE 因retry_at已被前一個(gè)改成租約時(shí)間而不滿足retry_atobj.retry_at條件更新 0 行即失敗——這就是原子租約防重入的原理。claim_processing_lock(lock_seconds60)實(shí)例方法版認(rèn)領(lǐng)。額外檢查終態(tài)與retry_at is None通過(guò)后委托給claim_for_worker。runner 處理每個(gè)任務(wù)前都調(diào)用它拿到鎖才繼續(xù)執(zhí)行。安全更新safe_update / update_and_requeuedef safe_update(self, update_fields, *, refreshTrue, extra_filterNone) - bool: values dict(update_fields) values.setdefault(modified_at, timezone.now()) queryset type(self).objects.filter(pkself.pk) if extra_filter: queryset queryset.filter(**extra_filter) updated queryset.update(**values) if updated ! 1 and extra_filter: current type(self).objects.filter(pkself.pk).values(status).first() logger.info( SafeUpdateGuardMiss: %s row %s extra_filter%s current_status%s loaded_status%s ..., ... ) if refresh: try: self.refresh_from_db() except type(self).DoesNotExist: pass return updated 1safe_update()用帶可選extra_filter的 UPDATE 實(shí)現(xiàn)帶守衛(wèi)的原子更新返回是否恰好更新 1 行。守衛(wèi)失敗時(shí)如extra_filter指定的狀態(tài)已被他人改變會(huì)記錄SafeUpdateGuardMiss日志便于排查競(jìng)態(tài)。默認(rèn)在更新后refresh_from_db()同步內(nèi)存態(tài)。update_and_requeue(**kwargs)safe_update的特化版本固定extra_filter{retry_at: self.retry_at}——即只有租約未變時(shí)才能更新再次確認(rèn)所有權(quán)后才允許改寫(xiě)狀態(tài)。runner 中大量使用它推進(jìn) Crawl/Snapshot 的調(diào)度狀態(tài)見(jiàn) runner.py。save() 覆蓋runner 進(jìn)程外的保存告警ModelWithQueue.save()覆寫(xiě)了 Django 的save()實(shí)現(xiàn)了一個(gè)運(yùn)行期安全網(wǎng)當(dāng)非新增行在非 ORCHESTRATOR編排器進(jìn)程中被save()時(shí)會(huì)通過(guò)棧幀回溯利用MODULE_PATH/PACKAGE_ROOT/REPO_ROOT定位調(diào)用者跳過(guò)本模塊與 site-packages 幀記錄警告日志指明調(diào)用方文件與行號(hào)。其意圖是隊(duì)列行的狀態(tài)變更應(yīng)當(dāng)由 runner 統(tǒng)一管理直接在任意代碼里save()可能繞過(guò)租約守衛(wèi)造成狀態(tài)錯(cuò)亂該告警幫助開(kāi)發(fā)者盡早發(fā)現(xiàn)這類(lèi)越權(quán)寫(xiě)庫(kù)。子類(lèi)可通過(guò)warn_on_save_outside_runner False關(guān)閉Binary 模型即如此因?yàn)樗葿inaryService同步驅(qū)動(dòng)安裝流程見(jiàn) machine/models.py。輔助方法status_counts / extend_choices / StatusField / RetryAtFieldstatus_counts(querysetNone, statusesNone) - dict[str, int]類(lèi)方法按狀態(tài)統(tǒng)計(jì)行數(shù)返回如{queued: 12, started: 3, ...}用于調(diào)度概覽與 admin 面板展示。extend_choices(base_choices)一個(gè)類(lèi)裝飾器工廠把基礎(chǔ)枚舉與附加枚舉合并生成新的TextChoices供需要擴(kuò)展?fàn)顟B(tài)集的模型使用。StatusField(**kwargs)/RetryAtField(**kwargs)類(lèi)方法字段工廠以默認(rèn)字段定義為基礎(chǔ)合并覆蓋參數(shù)子類(lèi)聲明字段時(shí)直接復(fù)用Crawl、Snapshot、Binary 三處均如此。子類(lèi)化實(shí)踐Crawl、Snapshot 與 Binary從倉(cāng)庫(kù)看ModelWithQueue被三個(gè)模型繼承展示了默認(rèn)協(xié)議復(fù)用 按需定制的完整模式Crawl 與 Snapshot直接復(fù)用默認(rèn)四態(tài)Crawl 與 Snapshot 都按默認(rèn)協(xié)議定制# crawls/models.py class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWithQueue): status ModelWithQueue.StatusField( choicesModelWithQueue.StatusChoices, defaultModelWithQueue.StatusChoices.QUEUED, ) retry_at ModelWithQueue.RetryAtField(defaulttimezone.now) StatusChoices ModelWithQueue.StatusChoices INITIAL_STATE StatusChoices.QUEUED ACTIVE_STATE StatusChoices.STARTED FINAL_STATES (StatusChoices.SEALED,) FINAL_OR_ACTIVE_STATES (*FINAL_STATES, ACTIVE_STATE)Snapshot 結(jié)構(gòu)相同core/models.py并額外定義RUNNABLE_STATES (QUEUED, STARTED)、OPEN_STATES等派生集合。兩者共享默認(rèn)四態(tài)queued → started → sealed可被paused中斷。Binary完全重定義狀態(tài)機(jī)Binary 演示了最激進(jìn)的定制——它只用兩態(tài)class StatusChoices(models.TextChoices): QUEUED queued, Queued INSTALLED installed, Installed status ModelWithQueue.StatusField(choicesStatusChoices.choices, defaultStatusChoices.QUEUED, max_length16) INITIAL_STATE StatusChoices.QUEUED ACTIVE_STATE StatusChoices.QUEUED # 活躍態(tài)即 QUEUED FINAL_STATES (StatusChoices.INSTALLED,) warn_on_save_outside_runner FalseBinary 的模塊 docstring 說(shuō)明了設(shè)計(jì)意圖安裝流程是同步的queued → installed轉(zhuǎn)換失敗時(shí)保持queued并把retry_at順延以便稍后重試數(shù)據(jù)庫(kù)行是唯一的生命周期狀態(tài)worker 中斷后無(wú)需在內(nèi)存中二次協(xié)調(diào)狀態(tài)機(jī)。它還演示了update_and_requeue(retry_atNone, statusINSTALLED)的收尾用法machine/models.py。運(yùn)行器中的真實(shí)調(diào)用鏈runnerservices/runner.py是把這套協(xié)議串起來(lái)的消費(fèi)端其處理每個(gè)到期任務(wù)的標(biāo)準(zhǔn)流程是選行通過(guò)retry_at__ltenow類(lèi)查詢(xún)等價(jià)于get_queue取出到期任務(wù)。認(rèn)領(lǐng)調(diào)用crawl.claim_processing_lock(lock_secondslock_seconds)L1361、L1395或Snapshot.claim_for_worker(snapshot, lock_secondslock_seconds)L1448原子獲取租約失敗即返回False說(shuō)明該行已被其他 worker 認(rèn)領(lǐng)。執(zhí)行與續(xù)租處理期間用crawl.update_and_requeue(statusSTARTED, retry_atnow timedelta(secondsACTIVE_STATE_LEASE_SECONDS))續(xù)租L1352把租約從 60 秒起步按需延長(zhǎng)。收尾終態(tài)任務(wù)如SEALED的 Crawl在claim_for_worker成功后執(zhí)行cleanup_runtime()最后update_and_requeue(retry_atNone)清空調(diào)度時(shí)間使其退出隊(duì)列L1405-L1411。值得注意的邊界處理當(dāng) Snapshot 的租約已過(guò)期但關(guān)聯(lián)的abx-dl鉤子進(jìn)程仍在運(yùn)行process_set.filter(statusrunning)時(shí)runner 不會(huì)啟動(dòng)第二波處理而是update_and_requeue(retry_atnow lock_seconds)重新排隊(duì)L1471-L1479保留快照級(jí)的所有權(quán)邊界防止重復(fù)執(zhí)行。測(cè)試驗(yàn)證與可觀測(cè)性倉(cāng)庫(kù)測(cè)試對(duì)這套協(xié)議有直接覆蓋可作為理解佐證test_crawl_runner.py 斷言snapshot.claim_processing_lock(lock_seconds60) is True驗(yàn)證單 worker 場(chǎng)景下的認(rèn)領(lǐng)成功路徑同文件 L414 使用crawl.update_and_requeue(statusQUEUED, retry_attimezone.now())構(gòu)造調(diào)度場(chǎng)景驗(yàn)證重排隊(duì)邏輯。此外認(rèn)領(lǐng)失敗與守衛(wèi)未命中都會(huì)留下可觀測(cè)日志SafeUpdateGuardMisssafe_update 守衛(wèi)失敗以及save() outside runner process告警配合status_counts()的狀態(tài)分布統(tǒng)計(jì)可以在運(yùn)維層面快速定位任務(wù)堆積在哪一態(tài)、是否被越權(quán)改寫(xiě)。小結(jié)archivebox.workers.models用約 230 行代碼給出了一個(gè)干凈、可復(fù)用的數(shù)據(jù)庫(kù)級(jí)任務(wù)隊(duì)列協(xié)議以statusretry_at兩個(gè)索引字段為存儲(chǔ)以條件 UPDATE 影響行數(shù)判定實(shí)現(xiàn)原子認(rèn)領(lǐng)與安全更新以pause/resume與RETRY_AT_MAX實(shí)現(xiàn)暫停語(yǔ)義以ACTIVE_STATE_LEASE_SECONDS維持活躍任務(wù)租約。Crawl、Snapshot、Binary 通過(guò)繼承與覆蓋各取所需runner 則作為唯一合法的狀態(tài)遷移入口消費(fèi)隊(duì)列——這套mixin 管協(xié)議、模型管遷移、runner 管執(zhí)行的設(shè)計(jì)是 ArchiveBox 多進(jìn)程調(diào)度一致性的基石?！久赓M(fèi)下載鏈接】ArchiveBox Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...項(xiàng)目地址: https://gitcode.com/gh_mirrors/ar/ArchiveBox創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考