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

从0到1使用Job Iteration重构长时运行的Rails后台任务【免费下载链接】job-iterationMakes your background jobs interruptible and resumable by design.项目地址: https://gitcode.com/gh_mirrors/jo/job-iterationJob 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添加到你的Gemfilegem 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 endeach_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 end2. 处理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 end3. 嵌套迭代处理多层级数据时可以使用嵌套枚举器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支持的队列适配器与环境Job Iteration支持主流的队列适配器无需额外配置即可实现优雅中断ResqueSidekiqGoodJobSolid QueueAmazon SQSDelayed::Job环境要求Ruby 3.1及以上Rails 7.0及以上深入学习资源Iteration工作原理作业参数语义自定义枚举器任务限流通过Job Iteration你可以构建更健壮、更可靠的Rails后台任务轻松应对云环境的不确定性和数据增长的挑战。开始使用Job Iteration让你的后台任务具备企业级的可靠性【免费下载链接】job-iterationMakes your background jobs interruptible and resumable by design.项目地址: https://gitcode.com/gh_mirrors/jo/job-iteration创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考