访问官网
Github
文档
论文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 快速的运行时。
开发
请参阅 [开发 River]。
致谢
River 在很大程度上受到了我们多年来使用其他后台作业库的经验的启发，特别是：
感谢你们推动软件生态系统的发展。
