一笔作业如何走完 River 的一生?Go 后台任务生命周期的 6 个路标

一笔作业如何走完 River 的一生?Go 后台任务生命周期的 6 个路标

【免费下载链接】riverFast and reliable background jobs in Go项目地址: https://gitcode.com/gh_mirrors/river/river

River 是面向 Go 的后台作业处理系统,主打快和可靠。做后台任务最怕三个坑:进程一重启,内存队列就清空;多实例同时抢,同一笔作业跑了两次;作业卡死在运行态,无人过问。River 的 River 架构解析思路很朴素:把作业状态机整个放进数据库,状态变更全走事务,每个环节再配一个兜底服务。下面沿一笔作业从入队到被清理的完整链路,逐个过一遍各组件的角色。

路标一:作业如何入队?Client 把任务写进数据库

先说为什么。作业只有落库,才算真正存在。调用 Enqueue 后,Client 先做参数校验(队列名、优先级、唯一性约束),再借 Driver 把作业插入 river_job 表。默认状态是 available,立即可取;想延迟执行,就把 scheduled_at 设到将来,状态落在 scheduled。

最容易误读的一点:Client 本身不持有队列。"队列"只是数据库里的行,Client 是带配置的执行器外壳,并发数、超时、保留期都定义在 client.go 的 Config 里:

// client.go:Config 控制客户端行为 type Config struct { CancelledJobRetentionPeriod time.Duration // 终态作业保留时长 Queues map[string]QueueConfig // 各队列并发数 Workers *Workers // 已注册 Worker 表 }

投递端到此结束。这一站解决的问题:任务从"内存里的想法"变成"数据库里的记录",重启不丢,任意多实例共享同一本账。

路标二:台账与方言,Driver 如何隔离存储细节

Client 不直接碰数据库,中间隔着一层 Driver 接口,定义在 riverdriver/ 包。它把查作业、取作业、改状态、收通知这些动作抽象成 Executor 与 Listener,方言差异(占位符是 $1 还是 ?、数组列怎么写)全封在接口后面。

仓库内置三套实现:riverpgxv5 面向 PostgreSQL,riversqlite 面向 SQLite 系,riverdatabasesql 面向标准 database/sql。迁移 SQL 随驱动打包,初始化一条命令走完。

这一站解决的问题:核心调度逻辑与具体数据库解耦,换存储不用改业务代码。

路标三:作业如何被取走执行?Worker 车间的接单流程

作业上架后谁来取?Client 内部的 fetcher 按队列并发数持续拉取 available 作业,每取到一笔就建一个 work unit,按 Kind 匹配到对应 Worker,调用 Work 执行。

Worker 是极窄的接口,见 worker.go,业务方真正要写的只有一个函数:

// worker.go:Worker 接口摘录 type Worker[T JobArgs] interface { Work(ctx context.Context, job *Job[T]) error Timeout(job *Job[T]) time.Duration NextRetry(job *Job[T]) time.Time }

两个细节值得注意。其一,Work 拿到的 ctx 带超时(默认 1 分钟,可在 Worker 或 Client 层覆盖),Worker 必须在 select 里监听 ctx.Done(),否则优雅停机时只能杀进程。其二,返回 nil 即视为成功,返回 error 则进入重试判定。

这一站解决的问题:执行与调度彻底分离,业务代码不关心取货和记账,只关心"这件事怎么做"。

路标四:作业如何从等待走到重试?调度器与救援队 🔁

失败是常态,排班和救援都在 internal/maintenance/ 这组常驻服务里:

• JobScheduler每 5 秒扫一次,把到点的 scheduled、retryable 作业改成 available,并发通知让监听方立刻取货,不必干等下一轮轮询。
• 重试判定:失败后按 retrypolicy 计算下次时间,Worker 可用 NextRetry 覆盖;超过 MaxAttempts 落到 discarded,不再自动重试。
• JobRescuer是巡线保安:每 30 秒检查一遍,把卡在 running 超过阈值(默认 1 小时)的作业拉回 retryable 或直接弃置,专治"卡死作业永远挂账"。

多实例部署时,这些服务只有一个"班组长"在跑:elector 基于数据库做领导者选举,落选实例只干活不维护,避免重复调度与重复清理。

这一站解决的问题:到点自动上架、失败退避重试、卡死有人捞,三种故障模式各有对应角色。

路标五:作业如何走到终局并被清理?状态机与保洁服务 🧹

终态(completed / discarded)之后,生命周期还没完,保洁出场。全部状态定义在 rivertype/river_type.go:

// rivertype/river_type.go:作业状态机 JobStateAvailable JobState = "available" // 可被取走 JobStateScheduled JobState = "scheduled" // 到点上架 JobStateRunning JobState = "running" JobStateRetryable JobState = "retryable" JobStateCompleted JobState = "completed" JobStateDiscarded JobState = "discarded" // 放弃重试

job_cleaner 按保留期批量删除:completed 与 cancelled 默认留 24 小时,discarded 留 7 天,到期物理删除。保留期内随时可以查历史、手动重试。

这一站解决的问题:终态记录有人定期清,表不无限膨胀;保留期又留出了排查窗口。

所谓 River 生命周期机制,本质就是:把"作业状态机"放进数据库,把"谁、何时、改哪个状态"拆成一组可单独测试的服务。

上手三步:五分钟让第一笔作业跑通 🚀

1. 建库选驱动:git clone https://gitcode.com/gh_mirrors/river/river拿示例代码;按环境选 riverpgxv5 或 riversqlite,跑完内置迁移。
2. 写最小 Worker:定义 JobArgs 并实现 Kind(),内嵌 WorkerDefaults 实现 Work,AddWorker 注册,Config.Queues 配好并发。
3. Start 再 Enqueue:client.Start 后投递一笔作业,盯数据库看状态 available → running → completed,再等 JobCleaner 把它清掉——整条链路亲眼走完一遍。

【免费下载链接】riverFast and reliable background jobs in Go项目地址: https://gitcode.com/gh_mirrors/river/river