Project Icon

river

为Go和Postgres打造的高性能任务队列

River是基于Go和Postgres的任务队列系统,特点是高性能和事务性入队。它提供多队列管理、定期任务、批量插入等功能,并包含Web UI用于监控。系统设计简化了任务定义和执行,适用于需要可靠后台处理的应用。River有效解决了常见的分布式系统问题。

River 构建状态 Go 参考

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 在很大程度上受到了我们多年来使用其他后台作业库的经验的启发,特别是:

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

项目侧边栏1项目侧边栏2
推荐项目
Project Cover

豆包MarsCode

豆包 MarsCode 是一款革命性的编程助手,通过AI技术提供代码补全、单测生成、代码解释和智能问答等功能,支持100+编程语言,与主流编辑器无缝集成,显著提升开发效率和代码质量。

Project Cover

AI写歌

Suno AI是一个革命性的AI音乐创作平台,能在短短30秒内帮助用户创作出一首完整的歌曲。无论是寻找创作灵感还是需要快速制作音乐,Suno AI都是音乐爱好者和专业人士的理想选择。

Project Cover

有言AI

有言平台提供一站式AIGC视频创作解决方案,通过智能技术简化视频制作流程。无论是企业宣传还是个人分享,有言都能帮助用户快速、轻松地制作出专业级别的视频内容。

Project Cover

Kimi

Kimi AI助手提供多语言对话支持,能够阅读和理解用户上传的文件内容,解析网页信息,并结合搜索结果为用户提供详尽的答案。无论是日常咨询还是专业问题,Kimi都能以友好、专业的方式提供帮助。

Project Cover

阿里绘蛙

绘蛙是阿里巴巴集团推出的革命性AI电商营销平台。利用尖端人工智能技术,为商家提供一键生成商品图和营销文案的服务,显著提升内容创作效率和营销效果。适用于淘宝、天猫等电商平台,让商品第一时间被种草。

Project Cover

吐司

探索Tensor.Art平台的独特AI模型,免费访问各种图像生成与AI训练工具,从Stable Diffusion等基础模型开始,轻松实现创新图像生成。体验前沿的AI技术,推动个人和企业的创新发展。

Project Cover

SubCat字幕猫

SubCat字幕猫APP是一款创新的视频播放器,它将改变您观看视频的方式!SubCat结合了先进的人工智能技术,为您提供即时视频字幕翻译,无论是本地视频还是网络流媒体,让您轻松享受各种语言的内容。

Project Cover

美间AI

美间AI创意设计平台,利用前沿AI技术,为设计师和营销人员提供一站式设计解决方案。从智能海报到3D效果图,再到文案生成,美间让创意设计更简单、更高效。

Project Cover

AIWritePaper论文写作

AIWritePaper论文写作是一站式AI论文写作辅助工具,简化了选题、文献检索至论文撰写的整个过程。通过简单设定,平台可快速生成高质量论文大纲和全文,配合图表、参考文献等一应俱全,同时提供开题报告和答辩PPT等增值服务,保障数据安全,有效提升写作效率和论文质量。

投诉举报邮箱: service@vectorlightyear.com
@2024 懂AI·鲁ICP备2024100362号-6·鲁公网安备37021002001498号