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("排序后的字符串：%+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)
}

仅插入客户端

通常需要一个只用于插入作业而不处理作业的客户端。这可以通过省略 Queues 配置并跳过对 Start 的调用来实现：

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Workers: workers,
})
if err != nil {
    panic(err)
}

也可以省略 Workers，但最好包含它，这样 River 可以检查插入的作业类型是否有可以运行它们的工作器。

停止

客户端也应该在程序关闭时停止：

// 停止获取新工作并等待活跃作业完成。
if err := riverClient.Stop(ctx); err != nil {
    panic(err)
}

确保客户端干净地停止，同时又能及时停止，涉及一些复杂性。有关 River 停止模式的更多详细信息，请参阅[优雅关闭]。

插入作业

[Client.InsertTx] 与作业参数实例一起使用，在事务中插入要处理的作业：

_, err = riverClient.InsertTx(ctx, tx, SortArgs{
    Strings: []string{
        "鲸鱼", "老虎", "熊",
    },
}, nil)

if err != nil {
    panic(err)
}

完整代码请参阅 [InsertAndWork 示例]。

其他特性

• [批量作业插入]：使用 Postgres COPY FROM 高效地一次插入多个作业。

• [取消作业]：从工作函数内部取消作业。

• [错误和异常处理]。

• [多队列]：更好地保证作业吞吐量、工作器可用性和组件间隔离。

• [周期性和 cron 作业]。

• [定时作业]：在未来指定时间自动运行。

• [作业暂停]：从工作函数内部暂停作业。

• [订阅]：队列活动和统计信息，为日志记录和指标等遥测提供简单的钩子。

• [测试助手]：验证作业是否按预期插入。

• [事务性作业完成]：保证作业完成与事务中的其他更改一起提交。

• [唯一作业]：按参数、周期、队列和状态唯一。

• [Web UI]：用于检查和与作业和队列交互。

• [工作函数]：简化的工作器实现。

跨语言入队

River 支持在一些非 Go 语言中插入作业，然后由 Go 实现来处理。在性能敏感的情况下，这可能是理想的选择，因为作业可以利用 Go 快速的运行时。

• 从 Python 插入作业。
• 从 Ruby 插入作业。

开发

请参阅 [开发 River]。

致谢

River 在很大程度上受到了我们多年来使用其他后台作业库的经验的启发，特别是：

• Elixir 中的 Oban。
• Ruby 中的 Que、Sidekiq、Delayed::Job 和 GoodJob。
• .NET 中的 Hangfire。

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