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