active-job

Rails Active Job 专家

使用 Active Job 和 Solid Queue（Rails 8 的默认队列）创建健壮、经过充分测试的后台任务。

理念

核心原则：

1. Solid Queue 是默认选择 — Rails 8+ 内置 Solid Queue。使用它即可，无需依赖 Redis。
2. 任务必须具备幂等性 — 任务可能会运行多次，因此要据此设计。
3. 参数必须可序列化 — 使用通过 GlobalID 序列化的 ActiveRecord 对象、基本类型或自定义序列化器。绝不要传递任意 Ruby 对象。
4. 始终处理失败情况 — 每个任务都需要 retry_on 或 discard_on，不能例外。
5. 单独测试任务 — 使用 perform_enqueued_jobs 和 assert_enqueued_with 分别验证行为和入队情况。

何时使用此技能

• 创建新的后台任务
• 将耗时工作移出请求周期（邮件、API 调用、数据处理）
• 配置 Solid Queue（队列、工作进程、优先级、周期性任务）
• 处理任务重试、错误和死信
• 测试任务行为和入队情况
• 使用 deliver_later 集成 ActionMailer
• 设置批量入队、并发控制或任务延续

指南

步骤 1：生成任务

bin/rails generate job process_payment
# Creates:
#   app/jobs/process_payment_job.rb
#   test/jobs/process_payment_job_test.rb

指定队列：

bin/rails generate job process_payment --queue critical

步骤 2：定义任务

class ProcessPaymentJob < ApplicationJob
  queue_as :default

  retry_on Net::OpenTimeout, wait: :polynomially_longer, attempts: 5
  discard_on ActiveJob::DeserializationError

  def perform(order)
    # `order` is an ActiveRecord object — serialized via GlobalID automatically
    PaymentGateway.charge(order.amount, order.payment_method)
    order.update!(paid_at: Time.current)
  end
end

每个任务都必须包含：

• queue_as — 显式指定队列
• retry_on 和/或 discard_on — 错误处理策略
• 幂等的 perform — 可安全地多次运行

步骤 3：将任务入队

# Run as soon as a worker is free
ProcessPaymentJob.perform_later(order)

# Run at a specific time
ProcessPaymentJob.set(wait_until: Date.tomorrow.noon).perform_later(order)

# Run after a delay
ProcessPaymentJob.set(wait: 5.minutes).perform_later(order)

# Run with a specific priority (lower = higher priority)
ProcessPaymentJob.set(priority: 0).perform_later(order)

# Run synchronously (useful in console, tests, rake tasks)
ProcessPaymentJob.perform_now(order)

perform_later 与 perform_now：

• perform_later — 将任务加入后台队列。在控制器、模型和服务中使用。
• perform_now — 在当前进程中运行并阻塞。适用于控制台、rake 任务、测试，以及需要顺序执行的其他任务。

步骤 4：处理错误

务必定义错误处理。这一点不可妥协。

class ExternalApiJob < ApplicationJob
  # Retry transient failures with exponential backoff
  retry_on Net::OpenTimeout, wait: :polynomially_longer, attempts: 5
  retry_on Faraday::ConnectionError, wait: 30.seconds, attempts: 3

  # Discard permanently broken jobs
  discard_on ActiveJob::DeserializationError  # Record was deleted
  discard_on ArgumentError                     # Bad arguments, retrying won't help

  # Optional: callback when a job is discarded
  after_discard do |job, exception|
    Rails.logger.warn("Discarded #{job.class.name}: #{exception.message}")
    ErrorTracker.notify(exception, job_id: job.job_id)
  end

  def perform(record)
    # ...
  end
end

retry_on 选项：

• wait: — 等待秒数或 :polynomially_longer（默认值：3s）
• attempts: — 最大重试次数（默认值：5）
• queue: — 重试时将任务移至其他队列
• priority: — 重试时更改优先级
• jitter: — 为等待时间添加随机性（默认值：0.15 = 15%）

discard_on — 静默丢弃任务。用于不可恢复的错误。

步骤 5：配置队列和优先级

队列分配：

class ImportJob < ApplicationJob
  queue_as :low_priority
end

根据参数动态选择队列：

class ProcessVideoJob < ApplicationJob
  queue_as do
    video = self.arguments.first
    video.owner.premium? ? :premium : :default
  end
end

队列内的优先级（数字越小，优先级越高）：

class UrgentNotificationJob < ApplicationJob
  queue_as :notifications
  queue_with_priority 0  # Processed first within :notifications queue
end

class DigestNotificationJob < ApplicationJob
  queue_as :notifications
  queue_with_priority 10  # Processed after urgent ones
end

在 config/queue.yml 中配置 Solid Queue 队列顺序：

production:
  workers:
    - queues: [critical, default, low_priority]
      threads: 5
      polling_interval: 0.1

⚠️ 队列顺序优先于优先级。 无论各自的优先级值如何，critical 中的所有任务都会先于 default 中的任何任务处理。优先级只在单个队列内生效。

步骤 6：正确序列化参数

安全的参数类型（开箱即用）：

• 基本类型：String、Integer、Float、Boolean、nil、Symbol
• 时间类型：Date、Time、DateTime、ActiveSupport::TimeWithZone、Duration
• 集合：Array、Hash（仅限字符串或符号键）、Range
• 类：Module、Class
• ActiveRecord 对象 — 自动通过 GlobalID 序列化

危险 — 不要直接传递这些对象：

# ❌ WRONG — arbitrary Ruby objects are NOT serializable
SomeJob.perform_later(Struct.new(:a).new(1))
SomeJob.perform_later(OpenStruct.new(foo: "bar"))
SomeJob.perform_later(URI.parse("https://example.com"))
SomeJob.perform_later(Money.new(100, "USD"))  # Without custom serializer

# ✅ CORRECT — pass primitives, let the job reconstruct
SomeJob.perform_later("https://example.com")
SomeJob.perform_later(100, "USD")

# ✅ CORRECT — ActiveRecord objects use GlobalID
SomeJob.perform_later(user)  # Serialized as "gid://app/User/123"

为值对象创建自定义序列化器：

# app/serializers/money_serializer.rb
class MoneySerializer < ActiveJob::Serializers::ObjectSerializer
  def serialize(money)
    super("amount" => money.amount, "currency" => money.currency)
  end

  def deserialize(hash)
    Money.new(hash["amount"], hash["currency"])
  end

  def klass
    Money
  end
end

# config/initializers/custom_serializers.rb
Rails.application.config.active_job.custom_serializers << MoneySerializer

步骤 7：使用回调

class AuditedJob < ApplicationJob
  before_perform :log_start
  after_perform :log_completion
  around_perform :measure_duration

  # before_enqueue / around_enqueue / after_enqueue also available
  # after_discard — fires when discard_on matches

  private

  def log_start
    Rails.logger.info("Starting #{self.class.name} #{job_id}")
  end

  def log_completion
    Rails.logger.info("Completed #{self.class.name} #{job_id}")
  end

  def measure_duration
    started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    yield
    elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started
    StatsD.measure("jobs.#{self.class.name.underscore}.duration", elapsed)
  end
end

可用回调： before_enqueue、around_enqueue、after_enqueue、before_perform、around_perform、after_perform、after_discard

步骤 8：测试任务

测试任务行为（是否按预期执行？）：

require "test_helper"

class ProcessPaymentJobTest < ActiveSupport::TestCase
  test "marks order as paid" do
    order = orders(:unpaid_order)

    perform_enqueued_jobs do
      ProcessPaymentJob.perform_later(order)
    end

    order.reload
    assert order.paid_at.present?
  end

  test "retries on timeout" do
    order = orders(:unpaid_order)

    assert_enqueued_with(job: ProcessPaymentJob) do
      ProcessPaymentJob.perform_later(order)
    end
  end
end

测试任务是否成功入队（从调用代码中测试）：

class OrdersControllerTest < ActionDispatch::IntegrationTest
  test "enqueues payment job on create" do
    assert_enqueued_with(job: ProcessPaymentJob) do
      post orders_path, params: { order: valid_order_params }
    end
  end

  test "enqueues to correct queue" do
    assert_enqueued_with(job: ProcessPaymentJob, queue: "critical") do
      post orders_path, params: { order: valid_order_params }
    end
  end
end

测试辅助方法：

# Execute all enqueued jobs inline
perform_enqueued_jobs do
  SomeJob.perform_later(args)
end

# Execute only specific job types
perform_enqueued_jobs(only: ProcessPaymentJob) do
  # ...
end

# Assert a job was enqueued
assert_enqueued_with(job: MyJob, args: [record], queue: "default")

# Assert number of jobs enqueued
assert_enqueued_jobs 2 do
  # ...
end

# Assert no jobs enqueued
assert_no_enqueued_jobs do
  # ...
end

步骤 9：集成 ActionMailer

# Send email asynchronously — enqueues via Active Job
UserMailer.welcome(@user).deliver_later

# Send with delay
UserMailer.welcome(@user).deliver_later(wait: 1.hour)

# Send immediately (bypasses queue)
UserMailer.welcome(@user).deliver_now

⚠️ 在 Rake 任务中，始终使用 deliver_now — deliver_later 的任务可能尚未处理完，进程就已退出。

邮件任务的错误报告：

class ApplicationMailer < ActionMailer::Base
  ActionMailer::MailDeliveryJob.rescue_from(Exception) do |exception|
    Rails.error.report(exception)
    raise exception
  end
end

步骤 10：批量入队

# Create job instances (don't call perform_later individually)
jobs = users.map { |user| WelcomeEmailJob.new(user) }

# Enqueue all at once — single round-trip to the queue backend
ActiveJob.perform_all_later(jobs)

# With options
jobs = users.map { |user| WelcomeEmailJob.new(user).set(wait: 1.day) }
ActiveJob.perform_all_later(jobs)

# Mix different job classes
jobs = [
  CleanupJob.new(record),
  NotifyJob.new(user),
  ExportJob.new(data)
]
ActiveJob.perform_all_later(jobs)

⚠️ 批量入队时不会触发 around_enqueue 回调。 如果需要在批量操作中进行钩接，请使用带有 enqueue_all.active_job 事件的 ActiveSupport::Notifications。

Solid Queue 配置

请参阅 references/queues-and-priority.md，了解完整的多工作进程 queue.yml 示例。

启动队列工作进程

bin/jobs start  # Development — run in separate terminal

并发控制

class ImportJob < ApplicationJob
  # Only 2 imports per account at a time
  limits_concurrency to: 2, key: ->(account) { account }, duration: 5.minutes

  def perform(account)
    account.run_import
  end
end

定期任务（config/recurring.yml）

production:
  cleanup_old_records:
    class: CleanupJob
    schedule: every day at 3am
  send_daily_digest:
    class: DailyDigestJob
    args: [summary]
    schedule: every day at 9am
  health_check:
    command: "HealthCheck.run"
    schedule: every 5 minutes

作业续接（Rails 8.1+）

对于可能被中断的长时间运行作业：

class LargeImportJob < ApplicationJob
  include ActiveJob::Continuable

  def perform(import_id)
    @import = Import.find(import_id)

    step :validate do
      @import.validate_data!
    end

    step :process do |step|
      @import.rows.find_each(start: step.cursor) do |row|
        row.process!
        step.advance! from: row.id
      end
    end

    step :finalize do
      @import.mark_complete!
    end
  end
end

Agent 常见错误

1. 传递不可序列化的对象 — 仅支持基本类型、ActiveRecord（通过 GlobalID）和已注册的序列化器类型。其他所有类型都会引发 SerializationError。
2. 没有错误处理 — 每个作业都需要 retry_on 或 discard_on。没有错误处理的作业就是一颗定时炸弹。
3. 未处理 DeserializationError — 如果记录在入队和执行之间被删除，就会出现此错误。务必使用 discard_on ActiveJob::DeserializationError。
4. 没有超时的阻塞作业 — HTTP 调用和外部 API 调用必须设置超时。为超时错误添加 retry_on。
5. 非幂等作业 — 如果作业涉及收费，请添加防护条件：return if order.paid?。作业可能会运行两次。
6. 在生产环境的控制器中使用 perform_now — 这会阻塞请求。请使用 perform_later。
7. 在开发环境中忘记启动队列工作进程 — 执行 bin/jobs start，或在开发环境中配置 config.active_job.queue_adapter = :async。
8. 想测试入队却使用 perform_now — 使用 assert_enqueued_with 验证作业是否已入队，使用 perform_enqueued_jobs 验证行为。

ApplicationJob 基类

在 app/jobs/application_job.rb 中设置合理的默认值：

class ApplicationJob < ActiveJob::Base
  # Automatically retry transient failures
  retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3

  # Discard if the record no longer exists
  discard_on ActiveJob::DeserializationError

  # Report errors to your error tracker
  rescue_from(Exception) do |exception|
    Rails.error.report(exception)
    raise exception
  end

  # Wait for transaction commit before enqueuing (prevents race conditions)
  self.enqueue_after_transaction_commit = true
end

运行与调试

# Start queue worker
bin/jobs start

# Run a job from console
ProcessPaymentJob.perform_now(Order.last)

# Check queue status (with mission_control-jobs gem)
# Visit /jobs in browser

# Enable verbose enqueue logging
# config/environments/development.rb
config.active_job.verbose_enqueue_logs = true

参考文件

详细模式和边缘情况请参阅 references/ 目录：

• references/serialization.md — 作业生命周期、GlobalID、参数类型、自定义序列化器
• references/error-handling.md — retry_on、discard_on、幂等性防护
• references/queues-and-priority.md — 多工作进程配置、命名、优先级、ActionMailer 集成
• references/testing.md — 入队断言、作业行为、错误处理、定时作业测试
• references/solid-queue.md — 数据库配置、并发控制、定期任务、监控
• references/advanced-patterns.md — 续接、性能、反模式、迁移指南