Rails Active Job 實(shí)戰(zhàn)詳解作業(yè)聲明、延遲調(diào)度、隊(duì)列適配器與 Continuations 斷點(diǎn)續(xù)跑【免費(fèi)下載鏈接】railsRuby on Rails項(xiàng)目地址: https://gitcode.com/GitHub_Trending/rai/railsActive Job 是 Rails 的異步作業(yè)框架負(fù)責(zé)把稍后要執(zhí)行的活兒從 HTTP 請(qǐng)求-響應(yīng)周期中剝離出來(lái)交給各種隊(duì)列后端Queuing Backend執(zhí)行。本文基于 Rails 倉(cāng)庫(kù)中的 activejob/README.md 及其配套源碼展開完整覆蓋作業(yè)聲明與入隊(duì)、set調(diào)度選項(xiàng)、GlobalID 參數(shù)序列化、Action Mailer 的deliver_later集成、各隊(duì)列適配器的能力對(duì)比以及較新的 Continuations可中斷、可恢復(fù)作業(yè)機(jī)制并給出每個(gè)結(jié)論對(duì)應(yīng)的源碼與測(cè)試路徑方便你對(duì)照閱讀。一、Active Job 的定位與設(shè)計(jì)目標(biāo)README 對(duì) Active Job 的定義是一個(gè)用于聲明作業(yè)、并讓其在多種隊(duì)列后端上運(yùn)行的框架。這些作業(yè)可以是定時(shí)清理、計(jì)費(fèi)扣款、郵件群發(fā)——任何可以拆成小塊、可并行執(zhí)行的工作單元都適用。它承擔(dān)兩個(gè)層面的職責(zé)統(tǒng)一作業(yè)接口。README 指出Active Job 的核心目標(biāo)是確保每個(gè) Rails 應(yīng)用都有一套作業(yè)基礎(chǔ)設(shè)施哪怕只是一個(gè) immediate runner立即執(zhí)行器。有了這層抽象框架特性和第三方 gem 就無(wú)需關(guān)心 Delayed Job 與 Resque 之間的 API 差異可以直接構(gòu)建在perform_later之上。后端可替換。選擇哪個(gè)隊(duì)列后端變成純粹的運(yùn)維決策operational concern切換后端時(shí)不需要重寫你的作業(yè)without having to rewrite your jobs。從源碼結(jié)構(gòu)看這一抽象的核心是 ActiveJob::Base它通過(guò)組合多個(gè)模塊Core、QueueAdapter、QueueName、QueuePriority、Enqueuing、Execution、Callbacks、Exceptions、Instrumentation、Logging等拼裝出完整的作業(yè)行為作業(yè)類只需繼承它并實(shí)現(xiàn)perform方法。README 給出的標(biāo)準(zhǔn)用法即class MyJob ActiveJob::Base queue_as :my_jobs def perform(record) record.do_work end end其中queue_as :my_jobs指定該作業(yè)進(jìn)入名為my_jobs的隊(duì)列其實(shí)現(xiàn)見(jiàn) QueueName 模塊除靜態(tài)名稱外還支持傳入 block 做動(dòng)態(tài)隊(duì)列名例如根據(jù)arguments.first的付費(fèi)狀態(tài)路由到:paid_feeds或:feeds并可通過(guò)queue_name_prefix/queue_name_delimiter默認(rèn)_為所有隊(duì)列加統(tǒng)一前綴便于在多應(yīng)用共享同一隊(duì)列中間件時(shí)做隔離。二、入隊(duì)與調(diào)度perform_later 及 set 選項(xiàng)README 給出了三種最典型的入隊(duì)方式這里完整保留并逐一說(shuō)明MyJob.perform_later record # 入隊(duì)隊(duì)列系統(tǒng)空閑時(shí)盡快執(zhí)行MyJob.set(wait_until: Date.tomorrow.noon).perform_later(record) # 明天中午 12 點(diǎn)執(zhí)行MyJob.set(wait: 1.week).perform_later(record) # 從現(xiàn)在起 1 周后執(zhí)行2.1set支持的全部選項(xiàng)結(jié)合 ActiveJob::Core::ClassMethods#set 的 RDocset接受四個(gè)選項(xiàng)選項(xiàng)含義:wait延遲指定時(shí)長(zhǎng)后執(zhí)行如VideoJob.set(wait: 5.minutes).perform_later(Video.last):wait_until在指定時(shí)刻執(zhí)行如VideoJob.set(wait_until: Time.now.tomorrow):queue覆蓋類級(jí)隊(duì)列名如VideoJob.set(queue: :some_queue):priority指定優(yōu)先級(jí)數(shù)值越小優(yōu)先級(jí)越高多個(gè)選項(xiàng)可以組合例如VideoJob.set(queue: :some_queue, wait: 5.minutes, priority: 10).perform_later(Video.last)。set返回一個(gè) ConfiguredJob 預(yù)配置對(duì)象對(duì)實(shí)例而言Core#set 的實(shí)現(xiàn)很直白:wait換算成scheduled_at options[:wait].seconds.from_now:wait_until直接賦值scheduled_at:queue走queue_name_from_part:priority轉(zhuǎn)成整數(shù)。2.2 入隊(duì)鏈路從 perform_later 到適配器從源碼結(jié)構(gòu)看perform_later到真正入隊(duì)的調(diào)用鏈如下見(jiàn) Enqueuing 模塊類方法perform_later(...)通過(guò)job_or_instantiate拿到作業(yè)實(shí)例若傳入的已是本類實(shí)例則直接復(fù)用隨后調(diào)用job.enqueue支持 yield 作業(yè)給可選 block返回入隊(duì)結(jié)果實(shí)例方法enqueue(options {})先執(zhí)行set(options)應(yīng)用調(diào)度選項(xiàng)再調(diào)用raw_enqueueraw_enqueue包裹:enqueue回調(diào)鏈執(zhí)行_raw_enqueue若設(shè)置了scheduled_at則走queue_adapter.enqueue_at(self, scheduled_at.to_f)延遲執(zhí)行否則走queue_adapter.enqueue(self)立即執(zhí)行若適配器拋出EnqueueError不會(huì)冒泡而是記錄到j(luò)ob.enqueue_error并把返回值置為false調(diào)用方可據(jù)此判斷入隊(duì)是否成功。一個(gè)容易被忽略但很實(shí)用的細(xì)節(jié)Enqueuing模塊定義了類屬性enqueue_after_transaction_commit默認(rèn)false。當(dāng) Active Job 與 Active Record 聯(lián)合使用時(shí)在數(shù)據(jù)庫(kù)事務(wù)內(nèi)調(diào)用perform_later會(huì)隱式把入隊(duì)推遲到事務(wù)提交之后回滾則丟棄作業(yè)避免作業(yè)先于數(shù)據(jù)可見(jiàn)的競(jìng)態(tài)?？稍谌只騿蝹€(gè)作業(yè)類上設(shè)置self.enqueue_after_transaction_commit true/false相關(guān)行為有專門的測(cè)試 enqueue_after_transaction_commit_test.rb 覆蓋。作業(yè)執(zhí)行完畢后ActiveJob::Core#serialize 決定了交給隊(duì)列后端的完整數(shù)據(jù)結(jié)構(gòu)job_class、job_id默認(rèn)SecureRandom.uuid、queue_name、priority、arguments序列化后的參數(shù)、executions、locale、timezone、enqueued_at、scheduled_at。這意味著作業(yè)的執(zhí)行環(huán)境時(shí)區(qū)、語(yǔ)言會(huì)隨作業(yè)數(shù)據(jù)一起跨進(jìn)程傳遞執(zhí)行端通過(guò)deserialize恢復(fù)。三、GlobalID 參數(shù)支持直接傳 Active Record 對(duì)象README 專門用一小節(jié)強(qiáng)調(diào)Active Job 支持對(duì)參數(shù)做 [GlobalID 序列化]這使得直接把活的 Active Record 對(duì)象傳給作業(yè)成為可能而不必像過(guò)去那樣傳類名 id再手動(dòng)constantize.find。README 給出的對(duì)比示例值得完整保留改造前class TrashableCleanupJob def perform(trashable_class, trashable_id, depth) trashable trashable_class.constantize.find(trashable_id) trashable.cleanup(depth) end end改造后class TrashableCleanupJob def perform(trashable, depth) trashable.cleanup(depth) end end其適用范圍是任何混入了GlobalID::Identification的類默認(rèn)包括所有 Active Record 模型。默認(rèn)情況下ActiveJob::Arguments 接受的原生類型包括String、Integer、Float、NilClass、TrueClass、FalseClass、BigDecimal、Symbol、Date、Time、DateTime、ActiveSupport::TimeWithZone、ActiveSupport::Duration、Hash、ActiveSupport::HashWithIndifferentAccess、Array、Range以及 GlobalID 實(shí)例這一白名單在 Enqueuing 的perform_later文檔 中有明確說(shuō)明且可以通過(guò)注冊(cè)自定義序列化器擴(kuò)展序列化邏輯見(jiàn) Serializers 模塊參數(shù)級(jí)測(cè)試見(jiàn) argument_serialization_test.rb。四、Action Mailer 的deliver_later把郵件變成作業(yè)README 指出 Active Job 還充當(dāng) Action Mailer#deliver_later的后端這讓任何郵件都能輕松變成一個(gè)稍后執(zhí)行的作業(yè)。README 認(rèn)為這是現(xiàn)代 Web 應(yīng)用中最常見(jiàn)的作業(yè)類型之一——把發(fā)信移到請(qǐng)求-響應(yīng)周期之外用戶就不必為它等待。在倉(cāng)庫(kù)中可以印證這條鏈路ActionMailer::Base 的文檔明確寫著NotifierMailer.welcome(User.first).deliver_later # enqueue the email sending to Active Job并提供了delivery_job默認(rèn)作業(yè)類與deliver_later_queue_name兩個(gè)類屬性用于定制。默認(rèn)的 ActionMailer::MailDeliveryJob 接收郵件序列化數(shù)據(jù)在后臺(tái)重新構(gòu)建并投遞從而完全復(fù)用 Active Job 的隊(duì)列、重試與調(diào)度能力。五、支持的隊(duì)列后端與適配器能力對(duì)比README 說(shuō)明 Active Job 內(nèi)置了多個(gè)隊(duì)列后端的適配器Resque、Delayed Job 等并給出了一條重要的治理聲明Rails 不再接收新適配器的 pull request正在把現(xiàn)有適配器向外抽取actively extracting the current adapters鼓勵(lì)庫(kù)作者在自家 gem 中或作為獨(dú)立 gem 提供 Active Job 適配器。從當(dāng)前倉(cāng)庫(kù) QueueAdapters 來(lái)看隨附的適配器包括外部隊(duì)列后端Backburner、Delayed Job、queue_classic、Resque、Sneakers以及三個(gè)用于測(cè)試和開發(fā)的內(nèi)置適配器AsyncAdapter線程池異步執(zhí)行、InlineAdapter進(jìn)程內(nèi)立即執(zhí)行即 README 所說(shuō)的 immediate runner、TestAdapter僅記錄不執(zhí)行供測(cè)試斷言。每個(gè)適配器位于 activejob/lib/active_job/queue_adapters/ 目錄下如 async_adapter.rb、inline_adapter.rb、test_adapter.rb、resque_adapter.rb 等。選擇后端時(shí)README 指向 ActiveJob::QueueAdapters 的文檔該文件中的能力對(duì)比表Backends Features值得直接引用列出了各后端在異步執(zhí)行Async、多隊(duì)列Queues、延遲執(zhí)行Delayed、優(yōu)先級(jí)Priorities、超時(shí)Timeout、重試Retries六個(gè)維度的支持情況后端AsyncQueuesDelayedPrioritiesTimeoutRetriesBackburnerYesYesYesYesJobGlobalDelayed JobYesYesYesJobGlobalGlobalqueue_classicYesYesYes*NoNoNoResqueYesYesYes (Gem)QueueGlobalYesSneakersYesYesNoQueueQueueNoActive Job AsyncYesYesYesNoNoNoActive Job InlineNoYesN/AN/AN/AN/AActive Job TestNoYesN/AN/AN/AN/A表中各維度的含義摘自同文件注釋Async作業(yè)能否以非阻塞方式運(yùn)行獨(dú)立/分叉進(jìn)程或不同線程N(yùn)o 表示作業(yè)在請(qǐng)求進(jìn)程內(nèi)同步運(yùn)行Queues能否用queue_as或set指定作業(yè)所在隊(duì)列Delayed能否通過(guò)perform_later在未來(lái)執(zhí)行標(biāo)注 (Gem) 表示需額外 gem 支持No 表示只能有機(jī)會(huì)就跑Priorities作業(yè)處理順序的控制粒度——Job作業(yè)級(jí)、Queue隊(duì)列級(jí)、Global全局配置、NoTimeout作業(yè)運(yùn)行時(shí)限的配置粒度作業(yè)級(jí) / 隊(duì)列級(jí) / 全局 / 不支持Retries重試次數(shù)配置能力作業(yè)級(jí) / 全局 / 不支持。補(bǔ)充說(shuō)明queue_classic 自 3.1 版起支持作業(yè)調(diào)度Delayed 列中的 Yes*舊版本可借助queue_classic-latergem。適配器的具體實(shí)現(xiàn)測(cè)試可參見(jiàn) adapter_test.rb、async_adapter_test.rb 等用例。六、Continuations可中斷、可恢復(fù)的長(zhǎng)作業(yè)README 在 Continuations 一節(jié)指出Continuations 允許作業(yè)被中斷并恢復(fù)interrupted and resumed詳見(jiàn)ActiveJob::Continuation。這是當(dāng)前倉(cāng)庫(kù)中一個(gè)內(nèi)容相當(dāng)豐富的新特性continuation.rb 的 RDoc 給出了完整的設(shè)計(jì)說(shuō)明其價(jià)值在于讓長(zhǎng)作業(yè)能在應(yīng)用重啟部署時(shí)保留進(jìn)度。核心機(jī)制如下6.1 基本用法step 定義步驟作業(yè)類include ActiveJob::Continuable后即啟用該能力被中斷的作業(yè)會(huì)自動(dòng)重試恢復(fù)。用step方法定義步驟步驟可帶可選 cursor游標(biāo)跟蹤進(jìn)度。官方示例摘自源碼 RDocclass ProcessImportJob ApplicationJob include ActiveJob::Continuable def perform(import_id) # 每次執(zhí)行含恢復(fù)都會(huì)運(yùn)行 import Import.find(import_id) step :validate do import.validate! end step(:process_records) do |step| import.records.find_each(start: step.cursor) do |record| record.process step.advance! from: record.id end end step :reprocess_records step :finalize end def reprocess_records(step) import.records.find_each(start: step.cursor) do |record| record.reprocess step.advance! from: record.id end end def finalize import.finalize! end end執(zhí)行語(yǔ)義是步驟按遇到的順序立即執(zhí)行作業(yè)被中斷后已完成的步驟會(huì)被跳過(guò)skip進(jìn)行中的步驟會(huì)從最后記錄的 cursor 處恢復(fù)不屬于任何 step 的代碼在每次運(yùn)行時(shí)都會(huì)執(zhí)行因此恢復(fù)場(chǎng)景下要注意冪等。step既可傳 block以 step 對(duì)象為參數(shù)也可傳方法名方法可不帶參數(shù)或接收 step 對(duì)象。6.2 Cursor游標(biāo)Cursor 用于在步驟內(nèi)跟蹤進(jìn)度可以是任何能經(jīng)ActiveJob::Base.serialize序列化的對(duì)象默認(rèn)nil恢復(fù)時(shí)自動(dòng)還原最后值由步驟代碼負(fù)責(zé)從正確位置繼續(xù)step.set!把 cursor 設(shè)為指定值示例items[step.cursor..].each { ...; step.set! (step.cursor || 0) 1 }定義步驟時(shí)可用start:指定初始游標(biāo)如step :iterate_items, start: 0step.advance!調(diào)用當(dāng)前游標(biāo)的succ前進(jìn)游標(biāo)不支持succ時(shí)拋ActiveJob::Continuation::UnadvanceableCursorErroradvance! from: record.id適用于 ID 不連續(xù)的場(chǎng)景如find_each按主鍵迭代游標(biāo)可以是數(shù)組以遍歷嵌套集合例如start: [0, 0]同時(shí)記錄(account_id, record_id)在兩層find_each中分別step.set!。6.3 Checkpoint檢查點(diǎn)與中斷時(shí)機(jī)設(shè)置/前進(jìn)游標(biāo)即自動(dòng)創(chuàng)建檢查點(diǎn)也可在無(wú)需更新游標(biāo)時(shí)手動(dòng)調(diào)用step.checkpoint!如循環(huán)中逐條destroy!并檢查點(diǎn)。檢查點(diǎn)的語(yǔ)義是作業(yè)可被中斷的位置屆時(shí)作業(yè)會(huì)調(diào)用queue_adapter.stopping?若返回true則以:stopping為原因拋ActiveJob::Continuation::Interrupt繼承自Exception而非StandardError因此不會(huì)被常規(guī)異常處理捕獲返回其他真值則作為中斷原因。每次作業(yè)執(zhí)行中除第一個(gè)步驟外每個(gè)步驟開始前都有自動(dòng)檢查點(diǎn)作業(yè)不會(huì)在適配器標(biāo)記 stopping 的瞬間被打斷而是繼續(xù)運(yùn)行到下一個(gè)檢查點(diǎn)或進(jìn)程停止——這意味著作業(yè)應(yīng)比關(guān)閉超時(shí)shutdown timeout更頻繁地打檢查點(diǎn)以保證優(yōu)雅重啟。被中斷后作業(yè)自動(dòng)重試進(jìn)度序列化在作業(yè)數(shù)據(jù)的continuation鍵下內(nèi)容為已完成步驟列表 當(dāng)前步驟及其游標(biāo)若步驟進(jìn)行中。6.4 Isolated Steps、Attributes 與錯(cuò)誤處理isolated: true讓某一步驟總是在獨(dú)立的一次執(zhí)行中運(yùn)行step :slow_step, isolated: true適合在作業(yè)寬限期內(nèi)無(wú)法打檢查點(diǎn)的超長(zhǎng)步驟——它確保步驟開始前進(jìn)度已序列化回作業(yè)數(shù)據(jù)Attributes步驟只保存序列化進(jìn)度、不保存其他狀態(tài)。跨步驟復(fù)用中間結(jié)果時(shí)用ActiveJob::Attributes聲明作業(yè)屬性該模塊已包含在Continuable中中斷時(shí)自動(dòng)序列化、恢復(fù)時(shí)自動(dòng)還原錯(cuò)誤自動(dòng)重試若作業(yè)在報(bào)錯(cuò)后本應(yīng)經(jīng) Active Job 重試而未重試進(jìn)度會(huì)交還給底層隊(duì)列后端而丟失。為緩解這一點(diǎn)只要作業(yè)已經(jīng)取得進(jìn)度完成過(guò)步驟或推進(jìn)過(guò)游標(biāo)報(bào)錯(cuò)時(shí)就會(huì)自動(dòng)重試。6.5 配置項(xiàng)ActiveJob::Continuable 提供三個(gè)類屬性配置均可在作業(yè)類中覆蓋配置默認(rèn)值說(shuō)明max_resumptionsnil無(wú)限次作業(yè)最多可恢復(fù)的次數(shù)從源碼看超限會(huì)拋Continuation::ResumeLimitErrorresume_options{ wait: 5.seconds }恢復(fù)時(shí)傳給retry_job的選項(xiàng)如{ wait: 1.seconds, queue: :resumed }resume_errors_after_advancingtrue推進(jìn)游標(biāo)之后發(fā)生錯(cuò)誤時(shí)是否仍恢復(fù)示例class ProcessImportJob ApplicationJob self.max_resumptions 3 self.resume_options { wait: 1.seconds, queue: :resumed } self.resume_errors_after_advancing false end行為測(cè)試可對(duì)照 continuation_test.rb 與 attributes_test.rb。七、安裝、許可證與延伸閱讀README 的安裝說(shuō)明最新版本可通過(guò) RubyGems 安裝gem install activejob源碼位于 Rails 項(xiàng)目的activejob目錄下本倉(cāng)庫(kù)即 activejob/。Active Job 以 MIT 許可證發(fā)布見(jiàn) activejob/MIT-LICENSE。更完整的入門內(nèi)容創(chuàng)建與入隊(duì)作業(yè)、配置后端、后臺(tái)執(zhí)行、異步發(fā)信、部署時(shí)暫停/恢復(fù)作業(yè)見(jiàn)倉(cāng)庫(kù)內(nèi)指南源文件 guides/source/active_job_basics.md——README 中指向的 Active Job Basics 指南在本倉(cāng)庫(kù)中對(duì)應(yīng)此文件適配器 API 的權(quán)威參考為 ActiveJob::QueueAdapters 的 RDoc。作業(yè)序列化的完整行為有 job_serialization_test.rb 與 serializers_test.rb 覆蓋入隊(duì)行為見(jiàn) queuing_test.rb。小結(jié)Active Job 通過(guò)perform_laterset(wait:/wait_until:/queue:/priority:)提供統(tǒng)一的入隊(duì)接口用 GlobalID 讓 Active Record 對(duì)象可作參數(shù)用適配器層把隊(duì)列后端的選型降維為運(yùn)維決策并以deliver_later承載了郵件發(fā)送這一最高頻的異步場(chǎng)景新增的 Continuations 則把長(zhǎng)作業(yè)跨部署存活這一運(yùn)維痛點(diǎn)納入了框架能力。理解了上述源碼路徑enqueuing.rb→core.rb→ 各queue_adapters/*與continuation.rb即可在本倉(cāng)庫(kù)中繼續(xù)深入任何一環(huán)的實(shí)現(xiàn)細(xì)節(jié)?！久赓M(fèi)下載鏈接】railsRuby on Rails項(xiàng)目地址: https://gitcode.com/GitHub_Trending/rai/rails創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考