riverqueue/river

来源：https://github.com/riverqueue/river

River 构建状态（https://github.com/riverqueue/river/actions） Go 参考（https://pkg.go.dev/github.com/riverqueue/river）

River 是一个用于 Go 和 Postgres 的健壮高性能任务处理系统。

参见首页、文档和GoDoc，以及 River UI 及其在线演示。

由于为 Postgres 构建，River 鼓励将同一数据库用于应用数据和任务队列。通过将任务入队与其他数据库变更事务性地结合，可以避免整类分布式系统问题。任务在其事务提交时保证入队，在事务回滚时被移除，并且在提交之前不可见。关于这一理念的更多背景，请参见事务性入队。

任务参数和工作者

任务通过结构体对定义，包含 JobArgs 的实现和 Worker 的实现。

任务参数包含 json 注解，定义了任务如何序列化到数据库及从数据库反序列化，以及一个“kind“（种类），这是一个稳定的字符串，用于唯一标识任务。

type SortArgs struct {
    // Strings 是一个待排序的字符串切片。
    Strings []string `json:"strings"`
}

func (SortArgs) Kind() string { return "sort" }

工作者暴露一个 Work 函数，规定了任务如何运行。

type SortWorker struct {
    // 嵌入的 WorkerDefaults 设置默认方法以满足 Worker 接口的其余部分：
    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
}

注册工作者

任务由其“kind“字符串唯一标识。工作者在启动时注册，以便 River 知道如何将任务分配给工作者：

workers := river.NewWorkers()
// AddWorker 在工作者已注册或无效时会引发 panic：
river.AddWorker(workers, &SortWorker{})

启动客户端

River Client 提供了任务插入的接口，并管理任务处理和维护服务。客户端通过数据库连接池、驱动和包含 Workers 包及其他设置的配置结构体创建。以下是一个 Client 处理一个队列（"default"），最多同时运行 100 个工作协程：

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)
}

// 内联运行客户端。所有执行的任务将继承 ctx：
if err := riverClient.Start(ctx); err != nil {
    panic(err)
}

也可以省略 Workers，但最好包含它，以便 River 能检查插入的任务种类是否有对应的工作者可以运行。

停止

客户端在程序关闭时也应停止。有多种方式可以实现（参见优雅关闭），但最简单的方法是当程序准备停止时取消传递给 Start 的上下文。例如，在收到 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()

// 停止获取新工作，等待活跃的任务完成。在 SoftStopTimeout 过期后取消任务。
if err := riverClient.Start(signalCtx); err != nil {
    panic(err)
}

<-riverClient.Stopped()

或者，显式调用 Stop：

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

仅插入客户端 可以插入任务但不处理它们，也不需要启动或停止。

插入任务

Client.InsertTx 与任务参数实例一起使用，在事务中插入一个任务：

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

if err != nil {
    panic(err)
}

参见 InsertAndWork 示例 获取完整代码。

其他特性

• 批量任务插入，高效地使用 Postgres COPY FROM 一次插入多个任务。
• 取消任务，从工作函数内部取消。
• 错误和 panic 处理。
• 多队列，更好地保证任务吞吐量、工作者可用性和组件隔离。
• 定时和 cron 任务。
• 计划任务，在未来的计划时间自动运行。
• 推迟任务，从工作函数内部推迟。
• 订阅 队列活动和统计信息，提供日志记录和指标等遥测的简易钩子。
• 测试辅助工具，验证任务是否按预期插入。
• 事务性任务完成，保证任务完成与事务中的其他变更一起提交。
• 唯一任务，按参数、周期、队列和状态去重。
• Web UI，用于检查和操作任务及队列。
• 工作函数，简化工作者实现。

跨语言入队

River 支持从某些非 Go 语言插入任务，然后由 Go 实现处理。这可能在性能敏感的场景下很有用，任务可以利用 Go 的快速运行时。

• 从 Python 插入任务（https://riverqueue.com/docs/python）。
• 从 Ruby 插入任务（https://riverqueue.com/docs/ruby）。

开发

参见开发 River。

致谢

River 在很大程度上受到我们多年来其他后台任务库经验的启发，最著名的是：

• Elixir 中的 Oban（https://github.com/sorentwo/oban）。
• Ruby 中的 Que（https://github.com/que-rb/que）、Sidekiq（https://github.com/sidekiq/sidekiq）、Delayed::Job（https://github.com/collectiveidea/delayed_job）和 GoodJob（https://github.com/bensheldon/good_job）。
• .NET 中的 Hangfire（https://www.hangfire.io/）。

感谢你们推动软件生态系统向前发展。