River for Rails：在 Rails 中通过 Active Job 接入 River 后台任务队列的完整指南

riverqueue-rails 是 River 多语言后台任务队列为 Rails 提供的官方集成 gem，它把 Active Job 无缝桥接到 River 已有的 Postgres / SQLite 运行时之上。本文将围绕这一集成，完整讲解从安装脚手架、配置初始化、专用 worker 进程部署，到事务原子性、重试与可观测性、测试策略的每一个环节，并给出仓库源码级的实现佐证，帮助你在 Rails 7.2 / 8.0 / 8.1 应用中落地一套由 Active Job 驱动、由 River 执行的后台任务体系。

一、集成概览：Active Job 与 River 的分工

riverqueue-rails（README.md）定位非常明确：集成（integration）而不是重写。它复用 River 核心 gem（Rails 无关）的客户端、调度、消费与维护机制，仅新增一层 Active Job 适配，让 Rails 开发者继续使用熟悉的 ApplicationJob、perform_later、retry_on 等 API，而底层行（row）的存储、领取、重试预算与队列并发全部由 River 承担。

从 gem 元数据（riverqueue-rails.gemspec）可以看到它与核心 gem 是解耦的：

• 支持 activejob >= 7.2, < 8.2 与 railties >= 7.2, < 8.2，即覆盖 Rails 7.2、8.0、8.1；
• 依赖 riverqueue-activerecord（版本与自身一致）作为 Active Record 驱动；
• 数据库 gem（pg 或 sqlite3）由应用自行选择，不在此 gem 中强制绑定。

集成后产生两类进程角色：

• Web 进程只负责插入（producer），绝不启动消费者；
• 专用 worker 进程负责消费，通过 river worker --rails 或生成的 bin/jobs start 启动。

这一分工贯穿全文，是理解后续所有配置与部署细节的出发点。

二、安装与项目脚手架

在 Gemfile 中加入 gem 并安装数据库驱动：

gem "riverqueue-rails"
gem "pg" # Or sqlite3.

随后执行三条命令完成脚手架与迁移：

bin/rails generate river:install
bin/rails river:migrate
bin/jobs start

生成器装了什么

bin/rails generate river:install 由 install_generator.rb 实现，它做两件事：

1. 把 river.rb 模板 复制到 config/initializers/river.rb；
2. 把 jobs 模板 复制为 bin/jobs 并赋予可执行权限。

生成的 bin/jobs 内容很短，核心只有两行：

require_relative "../config/environment"
require "riverqueue-rails"

abort "Usage: bin/jobs start" unless ARGV == ["start"]
exit River::Rails::Runner.start

它显式校验参数必须为 start，然后调用 River::Rails::Runner.start —— 也就是说 bin/jobs start 与 river worker --rails 走的是同一个 runner（见 runner.rb），差别只在于前者无 CLI 选项，而 --stop-timeout 等选项都属于 river worker。

迁移策略：复用 River 的规范 SQL

与常见的 Rails gem 不同，riverqueue-rails 不会创建独立的 Rails schema 迁移文件。bin/rails river:migrate 与 bin/rails river:status 这两个 rake 任务定义在 railtie.rb 中，内部直接使用核心 gem 的 River::Migrator，通过当前配置构建的 driver 执行官方打包的 Go SQL 迁移：

task migrate: :environment do
  River::Migrator.new(::Rails.application.config.river.build_driver).migrate
end

task status: :environment do
  River::Migrator.new(::Rails.application.config.river.build_driver).status.each do |migration|
    puts "#{migration.applied ? "applied" : "pending"} #{migration.version} #{migration.name}"
  end
end

两个任务都挂 :environment，因此会先完成 Rails 启动，再读取配置中的连接类。集成测试 integration_spec.rb 也验证了这两个任务可以正常执行并输出 applied 状态。

部署提示：迁移应在部署流程中显式执行一次，而不是在每个应用进程启动时执行；river:status 可用于核对迁移是否到位。

三、配置：初始化文件与懒执行语义

生成器提供的初始配置如下（river.rb 模板）：

Rails.application.configure do
  # Optional: select an abstract connection class by name, resolved after boot.
  # config.river.connection_class = "ApplicationRecord"
  config.active_job.queue_adapter = :river unless Rails.env.test?

  config.river.configure do
    River::Config.new(
      job_timeout: 300,
      logger: Rails.logger,
      queues: {default: 10, mailers: 5}
    )
  end
end

# For atomic enqueueing on the same Active Record connection, set
# self.enqueue_after_transaction_commit = false in ApplicationJob (Rails 8.x).
# Explicit after-commit deferral is respected, but is not an atomic SQL write.

config.river.configure 的懒执行

config.river 是一个 Configuration 实例。configure 方法只保存一个返回 River::Config 的 block，不会立即求值：

def configure(&block)
  raise ArgumentError, "configuration block required" unless block

  @factory = block
  @client = nil
end

从源码可见（configuration.rb），该 block 在两种时机被调用：

1. 应用 boot 完成后、懒加载首次请求 client 时；
2. 每次构造消费者（build_client）时。

因此 block 必须每次返回一个全新的 River::Config 实例，不能复用同一个对象。Configuration 还维护自己的默认值（configuration.rb）：连接类默认 "ActiveRecord::Base"、stop_timeout 默认 30 秒、默认队列 {"default" => 10}。测试 configuration_spec.rb 专门验证了"未请求 client 时 block 不会被调用"以及"block 必须存在"这两个语义。

集成做了哪些隐式工作

build_client（configuration.rb）在构造消费者时自动完成三件事：

1. 注册保留的 active_job worker：把 River::Rails::Worker 挂到 worker 注册表上（该 kind 名被保留，应用不能自行注册同名 worker，否则会抛 already registered 错误，见 configuration_spec.rb）；
2. 注入 Rails 执行边界插件：ExecutionPlugin 把每次任务执行包进 Rails.application.reloader.wrap，确保在开发模式 reload 与线程化上下文中清理连接和上下文；
3. 重包装周期性任务：periodic_jobs 的 constructor 也被套上 reloader.wrap。

队列与连接池的注意点

• queues: {default: 10, mailers: 5} 表示 default 队列每进程 10 个并发 worker、mailers 队列每进程 5 个，计数是按队列、按进程的，多个进程之间并发叠加；
• 必须把所有真实会用到的队列名都配置进去，包括 Active Job 队列前缀产生的队列和 Action Mailer 的 mailers 队列，否则对应任务不会被领取；
• 数据库连接池要为四类负载共同留足连接：应用自身的数据库操作、生产者插入、维护任务（如调度、清理）、消费者线程。

fork 与 Ractor 约束

Configuration#client（configuration.rb）维护一个进程内懒初始化的插入专用 client，并通过记录 @pid 实现 fork 后自动重建——fork 后第一次访问会因 PID 变化而丢弃旧 client、重新构建（对应测试 configuration_spec.rb 中 "rebuilds the insertion client after a fork" 用例）。也就是说，连接池的 fork 生命周期仍由 Rails / 进程服务器负责，集成只保证 client 本身安全。另外，client 不能跨 Ractor 共享，每个 Ractor 应使用自己的实例。

四、专用 worker 进程：river worker --rails

从 Rails 应用根目录启动消费者：

RAILS_ENV=production bundle exec river worker --rails
# Override config.river.stop_timeout for this process:
RAILS_ENV=production bundle exec river worker --rails --stop-timeout 60

river worker --rails 由 cli.rb 处理：它先 require config/environment.rb 完成 Rails boot，再加载 riverqueue-rails，最后调用 River::Rails::Runner.start(out:, stop_timeout:)。Runner（runner.rb）用 config.build_client 构建消费者并以 stop_timeout || config.stop_timeout 作为停止宽限。

关键约束：

• 不要在 web 进程的 initializer 里启动消费者，web 进程保持纯插入角色；
• 先迁移再启动消费者，避免消费端面对未就绪的 schema；
• 进程计数与重启交由外部 supervisor 管理——river worker 是一个前台进程，不守护化、不 fork 子进程、不自动补拉新进程，见 workers.md。

信号行为与优雅停机

WorkerRunner（worker_runner.rb）实现了完整的信号编排，三个信号各有语义：

信号	行为
TERM / INT	请求 stop，停止领取新任务并排空（drain）正在执行的任务，随后退出
TSTP	调用 stop(wait: false)：停止接收新工作，但进程保持存活，直到之后收到 TERM/INT 才完成退出
再次收到 stop 信号	跳过剩余宽限，直接中断正在执行的任务

时间线（worker_runner.rb）：

1. 收到 TERM/INT 后进入排空阶段，等待时长由 stop_timeout（默认 30 秒）决定；
2. 若宽限期届满仍有任务在执行，runner 会中断（interrupt）这些 attempt，再给 5 秒 finalization 时间（finalization_timeout: 5）；
3. 若 finalization 也超时，进程以 exit!(1) 强制退出，绕过 at_exit 处理器。

不同终止方式对任务状态的影响：

• 被成功完成中断的 attempt 会重新变为 available，可被再次领取，且不消耗 attempt 次数；
• 进程在 finalization 前意外死亡的任务，交给 River 常规的 stuck-job 恢复机制处理（当前仓库中恢复周期约为 1 小时）；
• 退出码语义：干净排空为 0；boot 失败、运行时线程故障、被中断的停止以及强制终止均为 1。

给 supervisor 的建议：终止宽限窗口应大于 stop_timeout + 5 秒 finalization + 一定余量；rolling 替换时先启动新实例并确认就绪，再向旧实例发 TERM，可先发 TSTP 让旧实例停止接收新工作。更完整的信号与滚动替换说明见 workers.md。

五、定义任务与发送邮件

集成后，任务定义完全沿用 Active Job 习惯：

class FulfillOrderJob < ApplicationJob
  queue_as :default
  retry_on PaymentGateway::Unavailable, attempts: 5, wait: 30.seconds
  discard_on ActiveJob::DeserializationError

  def perform(order)
    FulfillOrder.call(order)
  end
end

FulfillOrderJob.perform_later order
FulfillOrderJob.set(wait: 10.minutes).perform_later order
OrderMailer.receipt(order).deliver_later

Active Job 层完整接管了序列化、GlobalID 解析、回调、locale、时区以及自定义序列化器，应用代码无需感知 River 的存储格式。

批量插入走 River 原子批量

ActiveJob.perform_all_later 在适配器层通过 enqueue_all 实现，它调用 client.insert_many 一次性原子写入整批任务，成功后再逐一回填 provider_job_id 并标记 successfully_enqueued。集成测试 integration_spec.rb 验证了：批量中只要有一个非法任务（如非法优先级），整批回滚且不产生任何行。

优先级的取值边界

适配器对优先级做严格校验（river_adapter.rb）：

valid_priority = priority.nil? || (priority.is_a?(Integer) && (1..4).cover?(priority))
unless valid_priority
  raise ArgumentError, "River Active Job priority must be nil or an integer from 1 to 4"
end

即优先级必须是 nil（对应 River priority 1）或整数 1~4，超出范围直接抛 ArgumentError，绝不会静默改写语义。测试 rejects invalid priorities without silently clamping 与 does not partially insert a batch with an invalid job 均覆盖了这一行为。插入时还会写入 max_attempts: 25 与 scheduled_at（延迟任务），供崩溃恢复场景使用。

六、事务：连接类选择与原子性语义

选择连接类

默认连接类是 ActiveRecord::Base。若要改用其他抽象 Active Record 类，生产、消费、迁移三类操作会统一切换到该连接：

Rails.application.configure do
  config.river.connection_class = "ApplicationRecord"
end

这里有两个关键实现细节（configuration.rb）：

• 优先传类名字符串：connection_class 通过 constantize 懒解析，Rails 重载（reload）应用类后插入 client 会随之重建，避免持有旧的 reloadable 类对象；直接传类对象也支持；
• 修改连接类会使插入 client 失效：connection_class= 会清空缓存的 client，下一次访问重建。测试 connection_class_spec.rb 验证了切换连接类后 client 被重建，以及 reload 后按名称重新解析到新常量。

连接类自身的数据源通过常规的 establish_connection 或 connects_to 配置。改动连接配置后需要重启消费者才能生效。

不依赖 Rails 集成时直接选驱动

如果不在 Rails 应用中使用集成层，可以直接在驱动上指定连接类：

River::Driver::ActiveRecord.new(connection_class: ApplicationRecord)

原子性的正确姿势

只有在所选连接的事务内插入，业务写入与任务入队才能获得原子性。 仅仅共享同一个数据库 URL 是不够的——不同连接之间没有事务边界；为队列单独配置一个连接类/数据库虽然可行，但无法与应用写入所在的其他连接原子提交。

分片（shard）方面保持显式管理：插入跟随所选连接类的当前 Active Record role/shard；消费者需要各自合适的连接配置；请求内的 connected_to 块不会传播到 worker 线程。同一个驱动访问的所有 role/shard 应使用一致的数据库后端与 schema 布局；River 不提供自动分片发现或跨分片事务。

Rails 8 的 after-commit 语义

Rails 8.x 默认把入队延迟到事务提交后（enqueue_after_transaction_commit），这会破坏"同连接原子写入"的保证。若依赖原子性，需在应用侧显式关闭：

class ApplicationJob < ActiveJob::Base
  self.enqueue_after_transaction_commit = false
end

ApplicationRecord.transaction do
  order = Order.create!(status: "pending")
  FulfillOrderJob.perform_later(order)
end

版本差异（integration_spec.rb 中有对应验证）：

• Rails 7.2：适配器默认立即插入（river_adapter.rb 中 enqueue_after_transaction_commit? = false）；显式策略用 :never / :always（而非 Rails 8 的 false / true）；
• 所有受支持版本都会尊重应用显式设置的 after-commit 策略，但此时失去原子写入保证——返回的 provider_job_id 只能证明插入请求已返回，不能证明外层事务已提交；
• 任务完成与业务写入之间不会自动原子，需要应用自行设计补偿或幂等。

测试覆盖包括"事务内入队后回滚，行一并消失"、"显式 after-commit 策略下事务提交前无行、提交后有一行"以及"延迟任务在回滚时不入队"。

七、重试、中断与可观测性

谁拥有重试？

重试职责被刻意拆分为两层：

• Active Job 拥有应用级重试：retry_on 触发的重试会以同一个 Active Job UUID 插入新的 River 行，且新行的 provider_job_id 即为该行 ID 的字符串形式；重试被记录进原交付行的 metadata（active_job_outcome: "retried" 与 active_job_retry_id），见适配器的 record_retry（river_adapter.rb）；
• 未处理的错误（含 retry_on 耗尽后的丢弃）：当前行直接 discarded，不再叠加 River 重试轮次；
• 原生 River worker（非 Active Job）不受此影响，保留其常规重试策略；
• River 侧仍为每行保留 25 次 attempt 预算用于崩溃恢复，避免与 Active Job 的重试计数混淆；被卡住的 claim 由 River 常规维护（当前约 1 小时后）执行 stuck-job 恢复。

测试 integration_spec.rb 用 IntegrationRetryJob 验证了"两次执行共用一个 job_id、第一行标记 retried、两行状态分别为 completed 与 discarded"的完整链路。

River 控制流异常必须原样透传

River 的中断（Interrupted）、取消（JobCancelError）、暂停（JobSnoozeError）属于控制流异常，必须绕过 Active Job 的重试/丢弃处理器。为此 worker.rb 以 prepend 方式注入 ExecutionControl#rescue_with_handler（worker.rb）：

def rescue_with_handler(exception)
  if ActiveSupport::IsolatedExecutionState[:river_active_job] &&
      [River::ClientRuntime::Interrupted, River::JobCancelError, River::JobSnoozeError].any? { |type| exception.is_a?(type) }
    raise exception
  end

  super
end

这意味着即便任务里写了 retry_on StandardError 这种宽泛捕获，River 的取消/暂停/中断也不会被误当成应用错误处理。应用代码中应避免用宽泛的 rescue 块吞掉这些异常。 对应测试验证了：取消不会变成 Active Job 重试（行状态为 cancelled）、暂停不会触发重试（状态回到 scheduled 且 attempt 为 0）、stop 中断后任务回到 available。

交付完成的语义与可观测性

• 被 retry_on/discard_on 处理的任务，当前交付行以 completed 收尾，其 metadata 记录 active_job_outcome（retried 或 discarded）——"交付完成"不等于"逻辑任务成功"，观测时务必结合 outcome 判断；
• Active Job 的日志与通知机制保持不变，可继续接入原有观测体系；
• 每行在保留的 active_job kind 下存有版本化的 Active Job envelope（{"job" => ..., "version" => 1}）。Worker#work 会校验版本号并拒绝未知版本（worker.rb），测试 rejects unknown envelope versions 覆盖了这一点；
• 跨语言任务：Go 等其他语言可以共享同一 schema，但 active_job kind 的行必须由 Ruby 执行；跨语言协作应改用原生 River kind。

执行边界与一致性代价

需要认识到的两个非原子边界：

1. 重试入队与前序交付行的 finalization 不是原子的——崩溃可能造成重复，因此任务必须保持幂等；
2. Rails 的 executor/reloader 会包裹每次执行与周期性任务的 constructor，负责清理上下文与连接；Active Job 类在每次执行时重新解析（开发模式 reload 友好，测试 resolves Active Job classes again after Rails reloads 验证了这一点）。

另外，修改原生 River worker/plugin 注册或 periodic 配置后需要重启 worker，因为这些注册持有 Ruby 对象，不会随代码 reload 自动更新。

八、测试策略

测试分层建议（README.md 与 integration_spec.rb 共同支撑）：

• 普通应用测试：保留 Rails 的 :test adapter（即 ActiveJob::TestHelper），不引入真实数据库；这也解释了模板中 queue_adapter = :river unless Rails.env.test? 的用意；
• 持久化/运行时集成测试：改用 :river adapter，配合已迁移的隔离数据库；
• 线程化 worker 测试：需要关闭 transactional fixtures，否则事务隔离会掩盖真实的多线程行为；
• 仓库的 make test 已包含本 gem 在 Postgres 与 SQLite 上的测试；设置 RIVER_REQUIRE_DATABASES=1 make test 会让缺失 Postgres 环境直接判为失败（而非跳过）。

集成测试覆盖的典型场景包括：事务回滚后无残留行、延迟时间戳保留、批量插入与 provider ID 回填、批量中的非法任务整体回滚、Active Job 重试/丢弃/未处理错误的状态流转、GlobalID 反序列化与缺失记录丢弃、Action Mailer 投递、外部 claim 失败上报、模拟崩溃后的 stuck-job 恢复等。

九、脱离 Rails 应用使用适配器

集成层之外，适配器本身可以脱离 Rails 独立使用：实例化 ActiveJob::QueueAdapters::RiverAdapter.new(client:)（显式传入 client，见 river_adapter.rb），并在消费端把 River::Rails::Worker 注册到你的 River client 上。此时 Rails 相关的 boot、配置与执行边界包装（reloader 包裹、信号编排、rake 迁移任务）都由调用方自行负责——riverqueue-rails 的 Rails 集成正是基于这两块积木搭建起来的。

十、小结

riverqueue-rails 的价值在于让 Rails 团队以零学习成本获得 River 的生产级队列运行时：Active Job 负责序列化、重试与业务语义，River 负责存储、领取、并发、维护与优雅停机。把握住四个关键点即可稳定落地：web 进程只插入、worker 进程专职消费并交由 supervisor 管理；在同一连接的事务内入队以获得原子性（Rails 8 需显式关闭 after-commit 延迟）；River 的控制流异常不参与 Active Job 重试；重试入队与交付 finalization 非原子，任务务必幂等。按此模型接入，你的 Rails 应用就拥有了可横向扩展、可跨语言共享 schema 的后台任务基础设施。