企业官网建设流程全解析

• 任务调度
• 后端

导读

River 是一个以 Postgres 或 SQLite 为底层存储的快速、事务化任务队列（job queue），内置自动重试、定时调度、唯一任务与 Web UI 等能力，并同时提供 Go、JavaScript/TypeScript、Ruby、Rust 四套实现，各语言实现之间完全互通——在一个语言里入队、在另一个语言里处理是官方支持的一等公民用法。本文以 docs/README.md 为主线，结合仓库源码，系统讲解如何用 Go 定义任务参数与 Worker、注册并启动客户端、事务化插入任务，以及优雅关闭、多队列、周期任务等进阶特性，帮助你在一篇文章内掌握 River 的核心开发范式与底层实现原理。

一、River 是什么：用业务数据库承载任务队列

River 的核心定位是"使用与业务应用相同的数据库来承载任务队列"。这一哲学在项目自述中表述得非常明确：与其为任务队列单独引入 Redis、Kafka 等基础设施，River 鼓励把任务和业务数据放在同一数据库里，并通过事务保证两者的一致性。

这样做能够从根上规避一整类分布式系统难题：

• 任务保证随事务提交而入库：业务改动提交成功，任务就一定会存在；
• 任务随事务回滚而消失：业务回滚，任务也不会残留成为幽灵任务；
• 任务在提交之前不可见：事务未提交前，Worker 不会拿到该任务去执行，天然避免"任务先于数据可见"的竞态。

Client.InsertTx的源码注释进一步印证了这一机制：client.go 明确写道，把任务插入与业务变更放进同一事务，能保证任务启动时它所需的所有数据都已就绪；基于跨事务的快照可见性（snapshot visibility）保证，任务直到事务提交才会被工作，事务回滚则任务一并消失。

River 的完整能力清单包括：自动重试、定时调度、唯一任务、Web UI，以及 Go / JS/TS / Ruby / Rust 多语言实现。

二、Job Args 与 Worker：任务的定义方式

在 Go 中，一个任务由一对结构体共同定义：实现JobArgs接口的参数结构体 + 实现Worker接口的工人结构体。

2.1 JobArgs：任务参数与 Kind

JobArgs接口只有一个必须实现的方法Kind()，其定义在 job.go：

• 参数结构体通过json标签定义任务如何被序列化到数据库（River 底层以 JSON 存储 args 字段）；
• Kind()返回一个稳定字符串，用于唯一标识任务类型。

type SortArgs struct { // Strings is a slice of strings to sort. Strings []string `json:"strings"` } func (SortArgs) Kind() string { return "sort" }

关于 Kind 有几条来自源码的重要约束与建议（job.go）：

• Kind 应无空格，如my_custom_job、mycustomjob、my-custom-job；冒号、点、连字符、下划线等大多允许，但空格、逗号等会干扰 UI 功能的字符非法；
• 任务在数据库中以字符串 Kind 而非类型名标识，因此跨版本部署时可以重命名类型而不影响历史任务；
• 但 Kind 本身一经上线就不宜直接改名（除非清空数据库），否则旧 Kind 没有对应 Worker 处理。如需安全改名，可通过JobArgsWithKindAliases接口分三步完成（新 Kind 上线 → 旧 Kind 保留为别名 → 全部任务耗尽后移除别名），见 job.go。

2.2 Worker：任务的执行逻辑

Worker 通过Work函数规定任务如何执行。典型的实现是嵌入river.WorkerDefaults[T]，让默认方法补齐Worker接口的其余部分：

type SortWorker struct { // An embedded WorkerDefaults sets up default methods to fulfill the rest of // the Worker interface: river.WorkerDefaults[SortArgs] } func (w *SortWorker) Work(ctx context.Context, job *river.Job[SortArgs]) error { sort.Strings(job.Args.Strings) fmt.Printf("Sorted strings: %+v\n", job.Args.Strings) return nil }

Worker接口（worker.go）除Work外还有三个可重载的方法：

源码中特别强调（worker.go）：Work必须尊重 context 取消，才能配合客户端优雅关闭；凡是等待 channel、定时器或网络操作的任务，都应使用监听ctx.Done()的select。不尊重取消的任务，客户端超时或救援机制都拿它没办法，只能终止进程。

三、注册 Worker：NewWorkers 与 AddWorker

任务在数据库中以 Kind 唯一标识，因此客户端需要在启动前注册 Worker，以便把 Kind 映射到对应的执行者：

workers := river.NewWorkers() // AddWorker panics if the worker is already registered or invalid: river.AddWorker(workers, &SortWorker{})

AddWorker在 worker 已注册或配置非法时会panic（worker.go）。对大多数应用这是合理默认——谁也不希望在带着非法硬编码配置的情况下启动应用。若想避免 panic，可以使用返回 error 的AddWorkerSafely（worker.go）。

从源码实现（worker.go）可以看到，注册表Workers内部是一个map[string]workerInfo，以 Kind 为键；若任务参数实现了JobArgsWithKindAliases，其全部别名也会一并注册到同一 worker 信息下，且注册前会做完整性校验。

四、启动客户端：NewClient 与 Start

Client（client.go）提供任务插入接口，并负责管理任务处理与维护服务（maintenance services）。创建客户端需要三样东西：数据库连接池、驱动（driver）、以及包含Workers注册表和其他配置的Config结构体。

下面是一个同时处理默认队列"default"、最多 100 个并发 worker goroutine 的客户端示例：

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{ Queues: map[string]river.QueueConfig{ river.QueueDefault: {MaxWorkers: 100}, }, Workers: workers, }) if err != nil { panic(err) } // Run the client inline. All executed jobs will inherit from ctx: if err := riverClient.Start(ctx); err != nil { panic(err) }

4.1 驱动的选择

NewClient的第一个参数是类型参数化的 driver（client.go）。仓库为三种数据库访问方式提供了官方驱动：

• riverpgxv5：基于pgx/v5，事务类型为pgx.Tx；
• riverdatabasesql：基于database/sql，事务类型为*sql.Tx，可兼容多种底层实现（包括通过 pgx 的 stdlib 层连接 Postgres）；
• riversqlite：SQLite 驱动。

riverdatabasesql目录下有完整的可运行示例 example_client_from_context_database_sql_test.go，展示了sql.Open("pgx", ...)+riverdatabasesql.New(db)的用法。

4.2 Workers 字段为何建议必填

Workers也可以省略，但官方明确建议包含它——这样 River 能在插入任务时预先校验：被插入的 Kind 是否注册了可以运行它的 Worker（防止"忘了注册"这类低级事故）。同理，client.go 中的SkipUnknownJobCheck允许跳过这一检查，从而让一个客户端插入"由另一个客户端处理"的任务，行为上等同于 insert-only 客户端。

4.3 Start 的启动校验

从 Start 的实现 可以看到，启动时会先做若干硬性校验：未配置 Queues/Workers 的客户端不能 Start；Workers注册表为空会报错；随后还会对数据库做一次Ping 与驱动初始化，以便在客户端看似"已启动"但实际不可用时尽早返回错误。

4.4 Config 中值得关注的核心参数

Config结构体（client.go）字段众多，以下是 README 用法之外的高频配置项及其默认值：

其中QueueConfig.MaxWorkers的含义在源码中解释得很透彻（client.go）：它是单个客户端实例内该队列的最大并发数，不代表跨进程的总并发；要估算集群总并发，需要用它乘以运行同一数据库同一队列的节点数。取值范围 1～10,000。

五、优雅关闭：Stop / Stopped / SoftStopTimeout

客户端在程序退出时应当被停止。仓库提供了多种关闭途径（对应"graceful shutdown"），最简短的一种是：当程序准备停止时，取消传给Start的 context。例如监听SIGINT/SIGTERM信号并优雅退出：

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{ SoftStopTimeout: 10 * time.Second, ... }) if err != nil { panic(err) } signalCtx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM) defer stop() // Stop fetching new work and wait for active jobs to finish. Cancel jobs after // SoftStopTimeout elapses. if err := riverClient.Start(signalCtx); err != nil { panic(err) } <-riverClient.Stopped()

这段代码的语义在 client.go 中有精确定义：

1. 信号取消Start的 context，触发软停止（soft stop）：客户端停止抓取新任务，等待运行中的任务自然完成；
2. 若在SoftStopTimeout（此处为 10 秒）内任务仍未结束，客户端自动升级为硬停止——取消所有运行中任务的 context；
3. Stopped()返回的 channel 会在客户端完全停止后关闭（client.go），可用来等待关闭完成。

关于超时机制还有一个实现细节（client.go）：当配置了SoftStopTimeout时，River 会使用context.WithoutCancel把"工作 context"与"启动 context"分离，从而让启动 context 的取消先进入软停止阶段；若未配置超时，取消启动 context 等价于立即硬停止StopAndCancel。

也可以显式调用Stop完成关闭：

if err := riverClient.Stop(ctx); err != nil { panic(err) }

5.1 三种关闭方式对比

源码注释特别建议（client.go）：大多数情况下，配置SoftStopTimeout后使用Stop（或取消 Start context），优于直接调用StopAndCancel——前者给了任务先完成的机会，无需手动编排 Stop 与 StopAndCancel。

5.2 Insert-only 客户端

如果程序只负责入队、不处理任务，可以省略Queues配置，创建后无需 Start 也无需 Stop。这类 insert-only 客户端专门服务于"生产端与消费端分离"的架构。

六、插入任务：事务化入队 InsertTx

Client.InsertTx配合任务参数实例，可以在一个已有事务上插入任务：

_, err = riverClient.InsertTx(ctx, tx, SortArgs{ Strings: []string{ "whale", "tiger", "bear", }, }, nil) if err != nil { panic(err) }

注意此处的tx是调用方自己开启的事务（例如dbPool.Begin(ctx)），任务与业务数据在同一事务中提交——这正是第一节"事务化入队"哲学的落地 API。完整代码见 example_insert_and_work_test.go，其流程为：注册 Worker → 启动客户端 →Begin事务 →InsertTx插入 →Commit提交 → 收到任务完成的订阅事件 →Stop关闭，最终输出Sorted strings: [bear tiger whale]。

从实现角度看（client.go），InsertTx会把参数交给validateParamsAndInsertMany走统一的插入管线，并复用同一事务执行器（driver.UnwrapExecutor(tx)），因此任务插入与业务更新共享同一个数据库事务。

6.1 其他插入 API

Client还提供一组同族的插入方法：

• Insert(ctx, args, opts)：非事务插入，适合不需要与业务变更同事务的场景（源码实现位于 client.go，通过内部事务包裹保证单条插入原子性）；
• InsertMany(ctx, params)：批量插入多个任务，Postgres 驱动下走COPY FROM高效路径（README 中"Batch job insertion"即指此），示例见 example_batch_insert_test.go；
• InsertManyTx(ctx, tx, params)：批量插入的事务化版本，同样保证"提交前不可见、回滚则消失"（client.go）。

6.2 InsertOpts：插入期可用的选项

InsertTx的第三个参数是*InsertOpts，用于在插入时覆盖任务级与客户端级默认值。核心字段（insert_opts.go）：

其中UniqueOpts的维度矩阵语义（insert_opts.go）值得注意：只要任一维度被启用，任务的Kind 就自动计入唯一性（可用ExcludeKind排除）；ByArgs开启时整个编码后的 args 进入唯一性哈希，也可以在 JobArgs 字段上用river:"unique"标签只挑选部分字段参与哈希，避免无关字段破坏唯一性判断。

七、特性全景：README 之外值得掌握的能力

docs/README.md 列出了一系列特性，这里逐项给出它们在仓库中的落点，便于按图索骥：

两个容易被忽略的源码要点：

• 默认重试策略是指数退避：DefaultClientRetryPolicy按ATTEMPT^4计算下次重试（retry_policy.go），即第 1 次失败后 1 秒、第 2 次后 16 秒、第 3 次后 1 分 21 秒…… 且 snooze 不计入 attempt；
• 任务内可取出客户端：Work的 context 中携带当前Client，可用ClientFromContext取出（client_context.go），例如在任务内继续入队下游任务。

八、跨语言入队：多语言服务共享同一队列

任何 River 语言实现插入的任务，都可以被另一种语言实现处理——因此不同语言写的服务可以共享同一个任务队列。四种实现的源码分别位于仓库的 js、ruby、rust 目录（Go 实现在仓库根目录），各自的客户端与 Worker API 在对应目录的 README 与文档中有完整说明。

这一设计使"用 Go 写重活 Worker、用 Ruby 快速入队、用 TS 在 Node 侧提交任务"的异构架构成为可能，且各语言行为保持一致（如 Kind、重试、唯一性语义）。此外，River 也支持从 Python 侧直接插入任务。

九、本地试验与深入学习

• 想跑通完整流程，可直接阅读并参照 example_insert_and_work_test.go（事务化插入 + 处理）与 example_batch_insert_test.go（批量插入）；
• 驱动层事务语义可对比 riverpgxv5 与 riverdatabasesql 两个目录下的驱动实现与迁移 SQL（riverdriver/riverpgxv5/migration、riverdriver/riverdatabasesql/migration）；
• 客户端行为测试见 client_test.go，其中Test_Client_InsertTx（client_test.go）等用例覆盖了插入与事务语义；
• 参与或了解 River 自身开发流程，参见 docs/development.md。

结语

从定义JobArgs/Worker对、注册 Kind，到用NewClient配置队列与 Worker、用InsertTx与业务事务原子化入队，再到通过取消 Start context 配合SoftStopTimeout实现优雅关闭——River 的 Go 核心用法构成了一套完整且自洽的模型。配合多语言实现与丰富的内建特性，它把"任务队列"这一常见基础设施的运维心智降到最低：没有额外的中间件，没有独立的消息存储，所有状态都收敛在你已有的 Postgres 或 SQLite 之中。

• 任务调度
• 后端