当前位置：首页 > news >正文

从0到1：使用Job Iteration重构长时运行的Rails后台任务

news 2026/3/27 2:25:57

从0到1：使用Job Iteration重构长时运行的Rails后台任务

【免费下载链接】job-iterationMakes your background jobs interruptible and resumable by design.项目地址: https://gitcode.com/gh_mirrors/jo/job-iteration

Job Iteration是一个为ActiveJob设计的扩展，能让你的后台任务具备可中断和可恢复能力，通过设计保存任务的所有进度（也称为作业检查点）。对于处理大量数据或长时间运行的Rails后台任务，Job Iteration提供了可靠的解决方案，确保任务在面对部署、服务器重启或云环境中断时不会丢失进度。

为什么需要Job Iteration？

想象以下简单的Rails后台任务：

class SimpleJob < ApplicationJob def perform User.find_each do |user| user.notify_about_something end end end

当用户数量较少时，这个任务会很快完成。但随着数据增长，处理数百万用户可能需要数小时甚至数天。在频繁部署和 worker 重启的情况下，任务可能会丢失或从头开始，导致部分记录被重复处理。

云环境同样不可预测——AWS实例可能被标记为不健康需要重启，Kubernetes Pod可能被驱逐。这些情况都会导致任务进度丢失。Job Iteration正是为解决这些问题而设计，已在Shopify的生产环境中稳定运行多年。

快速开始：将普通任务转换为可中断任务

安装与配置

首先，将gem添加到你的Gemfile：

gem 'job-iteration'

执行安装命令：

bundle install

基本重构步骤

将传统的ActiveJob任务重构为Job Iteration任务只需两步：

包含JobIteration::Iteration模块
用build_enumerator和each_iteration方法替代perform方法

重构前：

class NotifyUsersJob < ApplicationJob def perform User.find_each do |user| user.notify_about_something end end end

重构后：

class NotifyUsersJob < ApplicationJob include JobIteration::Iteration def build_enumerator(cursor:) enumerator_builder.active_record_on_records( User.all, cursor: cursor, ) end def each_iteration(user) user.notify_about_something end end

each_iteration方法会为User.all关系中的每个用户调用，关系将按主键排序，与find_each行为一致。

进阶使用场景

1. 批处理记录

当需要处理大量记录时，可以使用批处理模式减少数据库查询次数：

class BatchesJob < ApplicationJob include JobIteration::Iteration def build_enumerator(product_id, cursor:) enumerator_builder.active_record_on_batches( Comment.where(product_id: product_id).select(:id), cursor: cursor, batch_size: 100, ) end def each_iteration(batch_of_comments, product_id) comment_ids = batch_of_comments.map(&:id) CommentService.call(comment_ids: comment_ids) end end

2. 处理CSV文件

Job Iteration提供了CSV枚举器，轻松处理大型CSV导入：

class CsvJob < ApplicationJob include JobIteration::Iteration def build_enumerator(import_id, cursor:) import = Import.find(import_id) enumerator_builder.csv(import.csv, cursor: cursor) end def each_iteration(csv_row, import_id) # 处理CSV行数据 end end

3. 嵌套迭代

处理多层级数据时，可以使用嵌套枚举器：

class NestedIterationJob < ApplicationJob include JobIteration::Iteration def build_enumerator(cursor:) enumerator_builder.nested( [ ->(cursor) { enumerator_builder.active_record_on_records(Shop.all, cursor: cursor) }, ->(shop, cursor) { enumerator_builder.active_record_on_records(shop.products, cursor: cursor) }, ->(_shop, product, cursor) { enumerator_builder.active_record_on_batch_relations(product.product_variants, cursor: cursor) } ], cursor: cursor ) end def each_iteration(product_variants_relation) # 处理产品变体数据 end end

最佳实践

单条记录处理优于批处理

尽管Job Iteration支持批处理，但单条记录处理允许更频繁的中断点，提高任务的可恢复性：

# 推荐：单条记录处理 class RecordsJob < ApplicationJob include JobIteration::Iteration def build_enumerator(product_id, cursor:) enumerator_builder.active_record_on_records( Comment.where(product_id: product_id), cursor: cursor, batch_size: 5, ) end def each_iteration(comment, product_id) comment.destroy end end

控制每次迭代时间

确保each_iteration方法执行时间不超过30秒，这有助于在收到中断信号时能够及时保存进度：

# config/initializers/instrumentation.rb ActiveSupport::Notifications.monotonic_subscribe("each_iteration.iteration") do |_, started, finished, _, tags| elapsed = finished - started if elapsed >= 25 # 25秒阈值 Rails.logger.warn "[Iteration] job_class=#{tags[:job_class]} 迭代时间超过限制" end end

设置最大作业运行时间

即使没有外部中断，也建议定期中断长时任务以释放worker资源：

# 全局设置 JobIteration.max_job_runtime = 5.minutes # 按任务设置 class MyJob < ApplicationJob include JobIteration::Iteration self.job_iteration_max_job_runtime = 3.minutes end