3 个版本 (破坏性更新)
0.19.0 | 2024 年 7 月 9 日 |
---|---|
0.18.0 | 2024 年 2 月 5 日 |
0.17.0 | 2024 年 1 月 18 日 |
#943 in 数据库接口
每月 135 次下载
在 background-jobs 中使用
91KB
1.5K SLoC
后台作业
此 crate 提供了在通常为同步的应用程序中异步运行某些过程的工具。标准示例是 Web 服务,其中某些事物需要被处理,但在用户等待浏览器响应时处理这些事物可能不是最佳体验。
用法
将后台作业添加到您的项目
[dependencies]
actix-rt = "2.2.0"
background-jobs = "0.15.0"
serde = { version = "1.0", features = ["derive"] }
要开始使用后台作业,首先应该定义一个作业。
作业是执行操作所需数据的组合以及该操作的逻辑。它们实现了 Job
,serde::Serialize
和 serde::DeserializeOwned
。
use background_jobs::{Job, BoxError};
use std::future::{ready, Ready};
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
pub struct MyJob {
some_usize: usize,
other_usize: usize,
}
impl MyJob {
pub fn new(some_usize: usize, other_usize: usize) -> Self {
MyJob {
some_usize,
other_usize,
}
}
}
impl Job for MyJob {
type State = ();
type Error = BoxError;
type Future = Ready<Result<(), BoxError>>;
const NAME: &'static str = "MyJob";
fn run(self, _: Self::State) -> Self::Future {
info!("args: {:?}", self);
ready(Ok(()))
}
}
作业的 run 方法还接受一个额外的参数,即作业期望使用的状态。应用程序中定义的所有作业的状态必须相同。默认情况下,状态是一个空元组,但您可能希望传递一些 Actix 地址或其他内容。
让我们重新定义作业以关注某些应用程序状态。
#[derive(Clone, Debug)]
pub struct MyState {
pub app_name: String,
}
impl MyState {
pub fn new(app_name: &str) -> Self {
MyState {
app_name: app_name.to_owned(),
}
}
}
impl Job for MyJob {
type State = MyState;
type Error = BoxError;
type Future = Ready<Result<(), BoxError>>;
// The name of the job. It is super important that each job has a unique name,
// because otherwise one job will overwrite another job when they're being
// registered.
const NAME: &'static str = "MyJob";
// The queue that this processor belongs to
//
// Workers have the option to subscribe to specific queues, so this is important to
// determine which worker will call the processor
//
// Jobs can optionally override the queue they're spawned on
const QUEUE: &'static str = DEFAULT_QUEUE;
// The number of times background-jobs should try to retry a job before giving up
//
// Jobs can optionally override this value
const MAX_RETRIES: MaxRetries = MaxRetries::Count(1);
// The logic to determine how often to retry this job if it fails
//
// Jobs can optionally override this value
const BACKOFF: Backoff = Backoff::Exponential(2);
fn run(self, state: Self::State) -> Self::Future {
info!("{}: args, {:?}", state.app_name, self);
ready(Ok(()))
}
}
运行作业
默认情况下,此 crate 随带 background-jobs-actix
功能启用。它使用 background-jobs-actix
crate 来启动服务器和工作进程,并提供创建新作业的机制。
background-jobs-actix
自身没有存储工作状态的机制。这可以通过手动实现 background-jobs-core
中的 Storage
trait 来实现,或者可以使用提供的内存存储。
在此之后,回到示例
主函数
use background_jobs::{create_server, actix::WorkerConfig, BoxError};
#[actix_rt::main]
async fn main() -> Result<(), BoxError> {
// Set up our Storage
// For this example, we use the default in-memory storage mechanism
use background_jobs::memory_storage::{ActixTimer, Storage};
let storage = Storage::new(ActixTimer);
// Configure and start our workers
let queue_handle = WorkerConfig::new(storage, move || MyState::new("My App"))
.register::<MyJob>()
.set_worker_count(DEFAULT_QUEUE, 16)
.start();
// Queue our jobs
queue_handle.queue(MyJob::new(1, 2))?;
queue_handle.queue(MyJob::new(3, 4))?;
queue_handle.queue(MyJob::new(5, 6))?;
// Block on Actix
actix_rt::signal::ctrl_c().await?;
Ok(())
}
完整示例
有关完整示例项目,请参阅 示例文件夹
提供自己的服务器/工作实现
如果您想基于此思路创建自己的作业处理器,可以依赖 background-jobs-core
crate,它提供 Job trait 以及用于实现作业处理器和作业存储的某些其他有用类型。
贡献
请随时为任何您发现的问题打开问题。请注意,任何贡献的代码都将根据 AGPLv3 许可。
许可证
版权所有 © 2022 Riley Trautman
background-jobs 是自由软件:您可以自由地重新分发和/或修改它,前提是您遵守自由软件基金会根据 GNU 通用公共许可证(GPL)发布的许可证条款,无论是许可证的第 3 版,还是(根据您的选择)任何更高版本的许可证。
background-jobs 分发时希望它有用,但没有任何保证;甚至没有对适销性或特定用途适用性的暗示性保证。有关详细信息,请参阅 GNU 通用公共许可证。此文件是 background-jobs 的一部分。
您应该已经随 background-jobs 收到了 GNU 通用公共许可证的副本。如果没有,请参阅 http://www.gnu.org/licenses/。
依赖项
~17–29MB
~423K SLoC