任务调度后端【免费下载链接】riverThe polyglot queue: Fast and reliable background jobs in Go, Ruby, Rust, and JS/TS on Postgres or SQLite.项目地址https://gitcode.com/gh_mirrors/river/river点击查看免费下载本文以仓库根目录的 rust/CHANGELOG.md 为核心骨架系统梳理 River Rust 客户端从 0.1.0 首次发布到 0.3.0 的版本演进逐条解读每个版本的破坏性变更、功能全景与设计意图并结合riverqueue系列 crate 的源码、README 与示例程序给出可直接复用的安装、迁移与调用示例。读完本文你将掌握 River Rust 五个 crate 的职责划分与版本策略、0.2.0/0.3.0 两次重命名背后的 API 设计取向以及首版发布所覆盖的完整功能清单与事务语义注意事项。一、版本发布总览与版本策略rust/CHANGELOG.md开篇即明确了版本管理规则riverqueue、riverqueue-macros、riverqueue-migrate、riverqueue-cli、riverqueue-test五个 workspace crate统一版本、统一发布格式遵循 Keep a Changelog两者独立记录、独立迭代。从当前仓库的 rust/Cargo.toml 可以看到workspace.package.version已升至0.3.0rust-version 1.95即 MSRVedition 为2024License 为 MPL-2.0。截至本文撰写时发布历史如下版本发布日期性质核心变更0.1.02026-10-06首次发布riverqueue全功能客户端上线五个 crate 成型0.2.02026-10-06破坏性重命名Client::start_with_graceful_shutdown→start_with_graceful_stop0.3.02026-10-07破坏性重命名WorkerRegistry→Workers、register→add、register_fn→add_fnUnreleased——当前工作区暂无已记录变更值得关注的是0.1.0 与 0.2.0 同在 2026-10-06 发布0.3.0 紧随其后于次日发布。CHANGELOG 中明确说明 0.2.0 的重命名是 a rare breaking name change as the new Rust API stabilizes——即趁 Rust API 尚在稳定期以少量破坏性变更加速收敛命名使其与 River for Go 保持一致。这正是理解两次重命名的关键语境。二、0.1.0River for Rust 首次发布0.1.0 是整份 CHANGELOG 信息量最大的条目它完整定义了 River Rust 客户端的初始能力边界。条目内容可拆解为三层主 crate 功能、配套 crates、事务语义约定。2.1riverqueue类型化、Tokio 驱动的客户端0.1.0 的核心交付是riverqueuecrate——一个类型化的、基于 Tokio 的客户端通过 SQLx 同时支持 Postgres 与 SQLite。它最重要的架构承诺是与 River for Go 共享数据库 schema 和作业协议job protocol因此 Rust 与 Go 的服务可以在同一个数据库里互相插入和处理对方创建的作业。这一点在多语言混合部署场景中尤为关键仓库中 rust/riverqueue/examples/mixed_go_rust.rs 正是这一能力的可运行示例。首版功能清单对应 CHANGELOG 0.1.0 条目原文包括类型化 worker通过#[derive(JobArgs)]声明可序列化的作业参数用异步函数或Worker实现处理逻辑事务化插入与完成插入与完成操作可挂接在调用方事务中执行unique / scheduled / periodic / resumable 作业去重、定时、周期与可恢复作业四大可靠性特性队列管理与作业取消暂停/恢复队列、取消/重试/删除持久化作业事件、hooks、中间件可观测与扩展机制leader 选举与维护服务由持锁 leader 运行调度、救援、清理等服务。riverqueue的安装方式来自 rust/riverqueue/README.md 与 rust/README.md[dependencies] riverqueue 0.3.0 serde { version 1, features [derive] } serde_json 1 tokio { version 1, features [macros, rt-multi-thread, signal] }其中三个 feature 开关决定了后端与扩展能力Feature默认作用postgres是通过 SQLx 使用 Postgressqlite否通过 SQLx 使用 SQLite 3.45 或更新版本chrono-tz否在 cron 的CRON_TZ/TZ前缀中支持 IANA 时区名如America/New_York若仅使用 SQLite应关闭默认特性riverqueue { version 0.3.0, default-features false, features [sqlite] }。River 不替调用方选择 SQLx 的 TLS 实现若数据库连接走 TLS需在自身 SQLx 依赖上启用其 TLS feature。API 中暴露 SQLx、Chrono、serde_json、tokio-util的类型均通过riverqueue::sqlx、riverqueue::chrono、riverqueue::serde_json、riverqueue::tokio_util再导出见 rust/riverqueue/src/lib.rs使用再导出或保持版本一致SQLx 0.9、Chrono 0.4、serde_json 1、tokio-util 0.7才能避免类型不匹配。2.2 快速开始首版 API 的完整形态结合 rust/riverqueue/README.md 的 Quick start0.1.0 建立起的客户端使用流程为定义可序列化参数 → 注册 worker → 执行迁移 → 启动客户端use riverqueue::migrate::PostgresMigrator; use riverqueue::sqlx::PgPool; use riverqueue::{ BoxError, Client, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, Deserialize, JobArgs, Serialize)] #[river(kind send_email)] struct SendEmail { address: String, } async fn send_email( context: WorkContext, job: JobSendEmail, ) - ResultWorkOutcome, BoxError { println!(sending email to {}, job.args.address); context.record_output(serde_json::json!({delivered: true}))?; Ok(WorkOutcome::Complete) } #[tokio::main] async fn main() - Result(), Boxdyn std::error::Error { let pool PgPool::connect(std::env::var(DATABASE_URL)?).await?; PostgresMigrator::new(pool.clone()).migrate_up().await?; let mut workers Workers::new(); workers.add_fn(send_email)?; let client Client::builder(pool) .workers(workers) .queue(default, QueueConfig::new(10)) .build()?; let mut run client.start_with_graceful_stop(async { let _ tokio::signal::ctrl_c().await; })?; client.insert(SendEmail { address: personexample.com.to_owned() }).await?; run.wait().await?; Ok(()) }注意示例中的Workers::new()与add_fn已经是 0.3.0 重命名后的写法在 0.1.0 时代对应的名称是WorkerRegistry与register_fn详见下文第四节。应用启动前必须先执行迁移生产环境可以把迁移作为独立部署步骤通过riverqueue-cli的riverqueue migrate-up命令完成。不启动客户端不带队列和 worker的客户端也可以直接用于插入作业——插入不要求客户端处于运行态。跨语言共享作业时作业 kind 与序列化后的 JSON 字段必须在各语言间保持一致。2.3 事务语义.tx(...)与 savepoint 约定0.1.0 条目末尾专门强调了一组事务约定这是使用时最容易踩坑的地方Requests run with.tx(...)use the callers transaction directly, without a savepoint or nested transaction, like River for Gos*Txmethods. Errors may leave partial writes, so roll back the transaction or create an explicit savepoint around the request.含义是凡是通过.tx(mut transaction)挂接事务的请求包括client.insert及client.jobs()、client.queues()上的各类操作都会直接在调用方事务内执行不会自动创建 savepoint 或嵌套事务。一旦请求返回错误事务中可能已经留下了部分写入因此正确的处理是回滚整个事务若希望从错误中恢复并继续使用该事务则需要调用方自行在请求前显式创建 savepoint。这与 River for Go 的*Tx方法语义完全一致。事务的启动方式也有讲究rust/riverqueue/README.md 建议使用riverqueue::database::begin_postgres(pool)Postgres或riverqueue::database::begin_sqlite_write(pool)SQLite。SQLite 版本使用BEGIN IMMEDIATE避免先读后写的事务因SQLITE_BUSY_SNAPSHOT失败。两者都在独立任务上开启中途放弃例如在select!或超时场景是安全的而 SQLx 自带的pool.begin()不具备这种安全性可能把仍处于事务中的连接归还给连接池。2.4 配套 crates宏、迁移、CLI 与测试工具0.1.0 同时交付了四个配套 crate各自职责如下Crate职责关键 APIriverqueue-macros过程宏#[derive(JobArgs)]riverqueue-migrate数据库迁移PostgresMigrator、SqliteMigratorriverqueue-cli命令行工具riverqueue migrate-up、riverqueue benchriverqueue-test测试辅助require_inserted*、TestJobBuilder、work_onceriverqueue-macros提供#[derive(JobArgs)]要求声明稳定的#[river(kind ...)]并可在属性中声明 kind 别名、默认队列、最大尝试次数、优先级、pending 状态、标签与默认唯一性配置完整语法见 rust/riverqueue-macros/README.md#[derive(Deserialize, JobArgs, Serialize)] #[river( kind send_invoice, queue billing, tags(billing), unique(by_args, by_period 24h, by_state(available, pending, running, scheduled)), insert_opts invoice_insert_opts, )] struct SendInvoice { #[river(unique)] invoice_id: i64, note: String, }unique(...)接受by_args、by_args(nested.path, ...)、by_period时长至少 1 秒如90s、1h30m、by_queue、by_state(...)与exclude_kind与UniqueOpts的 setter 一一对应。唯一路径遵循 Serde 序列化侧的rename/rename_all规则缺失的可选字段不参与唯一键计算flatten 或始终被跳过的唯一字段会被拒绝编译。riverqueue-migrate提供与 River for Go逐条一致的 Postgres 与 SQLite 迁移两种语言可以互迁同一个数据库。核心用法见 rust/riverqueue-migrate/README.mduse riverqueue_migrate::{Direction, MigrateOpts, PostgresMigrator, SqliteMigrator}; use sqlx::{PgPool, SqlitePool}; async fn migrate(postgres: PgPool, sqlite: SqlitePool) - Result(), Boxdyn std::error::Error { let migrator PostgresMigrator::new(postgres); let applied migrator.migrate_up().await?; println!(applied versions {applied:?}); let preview migrator .migrate(Direction::Down, MigrateOpts::new().with_dry_run(true)) .await?; for version in preview.versions { println!(would revert {:03} {}, version.version, version.name); } let validation SqliteMigrator::new(sqlite).validate(None).await?; if !validation.is_valid() { return Err(validation.to_string().into()); } Ok(()) }riverqueue-cli提供riverqueue命令migrate-up/migrate-down/migrate-list/validate管理 schemabench做基准测试会截断所选 River 作业表务必指向一次性数据库。命令示例cargo install riverqueue-cli riverqueue migrate-up --database-url postgres://localhost/app riverqueue migrate-up --database-url postgres://localhost/app --schema river riverqueue migrate-up --database-url sqlite://app.sqlite3 riverqueue bench --database-url postgres://localhost/river_bench --duration 30sriverqueue-test提供三类测试辅助见 rust/riverqueue-test/README.mdrequire_inserted/require_many_inserted/require_not_inserted系列断言_with变体支持按队列、优先级、状态、标签等属性断言_tx变体通过未提交事务读取TestJobBuilder构造真实的JobA并配合work_once脱离数据库单次调用 workerwork_with_client则是数据库回源版本可模拟 fetch 认领与生产环境调用链包括context.client()插入后续作业、context.job_complete_tx事务完成。需注意work_once不执行客户端 hooks、中间件、数据库事务、重试与完成持久化。三、0.2.0start_with_graceful_shutdown→start_with_graceful_stop0.2.0 只有一条变更但属于破坏性改动Breaking:RenamedClient::start_with_graceful_shutdowntoClient::start_with_graceful_stopto match Rivers standard start/stop terminology.将Client::start_with_graceful_shutdown重命名为start_with_graceful_stop以对齐 River 标准的 start/stop 术语体系。CHANGELOG 明确这是Rust API 稳定化过程中的一次罕见的破坏性改名。在 rust/riverqueue/src/client/run.rs 中可以确认新方法的实际签名pub fn start_with_graceful_stopF(self, signal: F) - ResultRunHandle, Error where F: std::future::FutureOutput () Send static,其语义对应 run.rs 的文档注释signalfuture 完成后等效于调用Stopper::stop()——客户端停止抓取新作业并让正在运行的作业自然结束若设置了soft_stop_timeout超时后会升级为取消。这与 Tokio 生态中 axum 的with_graceful_shutdown模式一致因此一个应用信号可以同时优雅关闭 HTTP 服务与 River 客户端CancellationToken也可以通过cancelled_owned()作为 signal 传入。围绕启动与停止RunHandle还提供了一组相关 API均可在 run.rs 找到start()L52普通启动stop_and_cancel()L553硬停止并取消运行中作业stopper()L680返回可克隆的Stopper供其他任务如信号处理调用detach()L635使客户端脱离监管继续运行。仓库示例 rust/riverqueue/examples/graceful_shutdown.rs 展示了完整模式第一次 Ctrl-C 触发stopper.stop()软停止第二次 Ctrl-C 触发stopper.stop_and_cancel()取消仍在运行的作业期间通过soft_stop_timeout控制升级时限。若使用 0.2.0 之前的 API把方法名改为start_with_graceful_stop即可完成升级。四、0.3.0WorkerRegistry→Workers0.3.0 同样是一条破坏性变更Breaking:RenamedWorkerRegistrytoWorkers,registertoadd, andregister_fntoadd_fnto align worker registration with Gos naming.将WorkerRegistry重命名为Workers、register重命名为add、register_fn重命名为add_fn目的是对齐 River for Go 的命名。两条方法都保留Result返回类型与原有注册行为——注册错误可以通过?向上传播。在 rust/riverqueue/src/worker.rs 与 rust/riverqueue/README.md 中可以看到重命名后的用法Workers::add注册一个实现Workertrait 的类型等价于 Go 的AddWorker。当某类作业需要自定义超时WorkerTimeout或自定义下次重试决策时应实现Worker。Workers::add_fn直接注册一个异步函数或捕获闭包把 Go 的WorkFunc适配器与注册合并为一次调用。两者均返回Result因此注册错误如重复 kind可立即通过?传播let mut workers Workers::new(); workers.add_fn(send_email)?; // 0.3.0 之前写作 workers.register_fn(send_email)?;升级到 0.3.0 时只需将WorkerRegistry替换为Workers、register替换为add、register_fn替换为add_fn其余注册行为不变。结合 0.2.0 的重命名可以看到一个清晰的收敛方向Rust 客户端在稳定化阶段主动向 Go 版 API 的命名靠拢从而降低两种语言混用时的认知负担。五、从 CHANGELOG 看项目演进与维护实践5.1 快速迭代背后的工程约束0.1.0 到 0.3.0 在两天内连续发布且两次都是破坏性变更这在已进入稳定期的开源项目里并不常见。CHANGELOG 的措辞给出了解释Rust API 尚处稳定化阶段as the new Rust API stabilizes项目选择在此窗口期内以最小成本完成命名收敛避免未来用户基数扩大后再被迫破坏性改名。对于读者而言如果正在跟随早期版本开发应优先升级到 0.3.0以锁定最终命名。5.2 发布与版本管理流程rust/docs/development.md 完整记录了 Rust 侧的发布流程与 CHANGELOG 中五个 crate 统一发布的约定相互印证版本号在 rust/Cargo.toml 的workspace.package.version统一维护Git 标签使用rust/vX.Y.Z与 Go module 标签彼此独立每次发布需同步更新rust/*/Cargo.toml中对其他 River crate 的精确版本依赖、README 中的版本示例并把rust/CHANGELOG.md的Unreleased条目移动到带日期的版本段发布顺序由 Cargo 处理依赖riverqueue-macros与riverqueue-migrate先于riverqueue随后是riverqueue-cli与riverqueue-testmake check/rust/package会构建五个 crate 的归档并验证各自能从打包源码独立构建但不会真正发布。这也解释了 CHANGELOG 当前的[Unreleased]节为空的原因任何新变更在正式发版前都会先登记在Unreleased发版时再归入带日期的版本段。5.3 变更日志中的 Go 互操作主线通读三个版本的条目可以发现一条贯穿始终的主线与 River for Go 的兼容与对齐。0.1.0 强调与 Go 共享 schema 与 job protocol0.2.0 与 0.3.0 的两次重命名都明确写着 to align with Gos naming。这条主线也体现在具体实现中riverqueue-macros的唯一键计算遵循 Serde 序列化命名规则以保持跨语言一致见 rust/riverqueue-macros/README.mdriverqueue的 encoding 模块 专门保证unique keys identical across Rust and Go而 rust/riverqueue/docs/mixed-deployments.md 与 rust/riverqueue/examples/mixed_go_rust.rs 则给出双语言共库部署的实践指引。因此对于同时维护 Go 与 Rust 服务的团队跟随 River Rust 客户端版本时尤其要注意升级到 0.3.0 之前服务端与 CI 中的 API 调用名Workers/add/add_fn/start_with_graceful_stop需要同步更新。结语从 0.1.0 的全功能首版、0.2.0 的启动/停止命名统一到 0.3.0 的 worker 注册命名统一rust/CHANGELOG.md记录的不只是三次发布更是一套 API 在设计窗口期内主动向 Go 侧对齐的演进轨迹。对使用者而言这几点值得牢记五个 crate 始终同版本发布升级时需整体看待.tx(...)请求直接运行在调用方事务中出错后必须回滚或自行兜底 savepointworker 注册与优雅停止的 API 已定型为Workers::add/add_fn与start_with_graceful_stop。后续新特性会先出现在[Unreleased]段持续关注该文件即可第一时间掌握 River Rust 客户端的演进方向。赞分享任务调度后端【免费下载链接】riverThe polyglot queue: Fast and reliable background jobs in Go, Ruby, Rust, and JS/TS on Postgres or SQLite.项目地址https://gitcode.com/gh_mirrors/river/river点击查看免费下载相关推荐Arco Design Affix 组件变更日志深度解读从版本演进看固钉的稳定化之路Arco Design Affix 组件变更日志深度解读从版本演进看固钉的稳定化之路 固钉Affix是 Arco Design React 组件库中用于将UI组件前端设计系统schedule 变更日志深度解读从 0.1.0 到 1.2.2 的 API 演进、行为变更与升级实践指南schedule 变更日志深度解读从 0.1.0 到 1.2.2 的 API 演进、行为变更与升级实践指南 本文以开源仓库 schedule Python任务调度后端Stylelint 变更日志深度解读从 0.1.0 到 17.14.1 的版本演进、破坏性变更与迁移实战Stylelint 变更日志深度解读从 0.1.0 到 17.14.1 的版本演进、破坏性变更与迁移实战 Stylelint 是一个帮助开发者避免 CSS 错代码质量静态分析前端上一篇【亲测免费】 Gource 使用指南可视化你的版本控制历史下一篇推荐文章【SymlinkCreator】打造你的高效文件链接管理新体验创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考