Fang

Rust的后台任务处理库。它可以使用PostgreSQL、SQLite或MySQL作为异步任务队列。

主要特性

以下是fang的一些主要特性

• 异步和线程化工作者。工作者可以以线程（线程化工作者）或tokio任务（异步工作者）的方式启动
• 计划任务。任务可以在未来的任何时间安排
• 周期性（CRON）任务。可以使用cron表达式安排任务
• 唯一任务。如果任务是唯一的，则不会在队列中重复
• 专用工作者。任务存储在单个表中，但工作者只能执行特定类型的任务
• 重试。任务可以以自定义退避模式重试

安装

1. 将其添加到您的Cargo.toml中

阻塞功能

[dependencies]
fang = { version = "0.11.0-rc1" , features = ["blocking"], default-features = false }

异步功能

• 将PostgreSQL作为队列

[dependencies]
fang = { version = "0.11.0-rc1" , features = ["asynk-postgres"], default-features = false }

• 将SQLite作为队列

[dependencies]
fang = { version = "0.11.0-rc1" , features = ["asynk-sqlite"], default-features = false }

• 将MySQL作为队列

[dependencies]
fang = { version = "0.11.0-rc1" , features = ["asynk-mysql"], default-features = false }

带有derive宏的异步功能

将database替换为您希望使用的后端。

[dependencies]
fang = { version = "0.11.0-rc1" , features = ["asynk-{database}", "derive-error" ], default-features = false }

所有功能

fang = { version = "0.11.0-rc1" }

支持rustc 1.77+

1. 在数据库中创建fang_tasks表。每个数据库的迁移可以在fang/{database}-migrations中找到，其中database是postgres、mysql或sqlite。

迁移也可以作为代码运行，导入特性migrations-{database}，其中database是要使用的后端队列。

[dependencies]
fang = { version = "0.11.0-rc1" , features = ["asynk-postgres", "migrations-postgres" ], default-features = false }

use fang::run_migrations_postgres;
run_migrations_postgres(&mut connection).unwrap();

使用方法

定义一个任务

阻塞功能

每个任务都应该实现由 fang::Runnable 特性，该特性被 fang 用于执行它。

如果您有 CustomError，建议实现 From<FangError>。这样，您就可以在 fang::Runnable 特性中的 run 函数内部使用 ? 操作符。

您可以使用宏 ToFangError 轻松实现它。此宏仅在功能 derive-error 中可用。

use fang::FangError;
use fang::Runnable;
use fang::typetag;
use fang::PgConnection;
use fang::serde::{Deserialize, Serialize};
use fang::ToFangError;
use std::fmt::Debug;


#[derive(Debug, ToFangError)]
enum CustomError {
    ErrorOne(String),
    ErrorTwo(u32),
}

fn my_func(num : u16) -> Result<(), CustomError> {
    if num == 0 {
        Err(CustomError::ErrorOne("is zero".to_string()))
    }

    if num > 500 {
        Err(CustomError::ErrorTwo(num))
    }

    Ok(())
}

#[derive(Serialize, Deserialize)]
#[serde(crate = "fang::serde")]
struct MyTask {
    pub number: u16,
}

#[typetag::serde]
impl Runnable for MyTask {
    fn run(&self, _queue: &dyn Queueable) -> Result<(), FangError> {
        println!("the number is {}", self.number);

        my_func(self.number)?;
        // You can use ? operator because
        // From<FangError> is implemented thanks to ToFangError derive macro.

        Ok(())
    }

    // If `uniq` is set to true and the task is already in the storage, it won't be inserted again
    // The existing record will be returned for for any insertions operaiton
    fn uniq(&self) -> bool {
        true
    }

    // This will be useful if you want to filter tasks.
    // the default value is `common`
    fn task_type(&self) -> String {
        "my_task".to_string()
    }

    // This will be useful if you would like to schedule tasks.
    // default value is None (the task is not scheduled, it's just executed as soon as it's inserted)
    fn cron(&self) -> Option<Scheduled> {
        let expression = "0/20 * * * Aug-Sep * 2022/1";
        Some(Scheduled::CronPattern(expression.to_string()))
    }

    // the maximum number of retries. Set it to 0 to make it not retriable
    // the default value is 20
    fn max_retries(&self) -> i32 {
        20
    }

    // backoff mode for retries
    fn backoff(&self, attempt: u32) -> u32 {
        u32::pow(2, attempt)
    }
}

如上例所示，特质的实现具有 #[typetag::serde] 属性，用于反序列化任务。

run 函数的第二个参数是一个实现了 fang::Queueable 的结构体。您可以重新使用它来操作任务队列，例如在当前任务执行期间添加新的作业。如果您不需要它，只需忽略它即可。

Asynk 功能

每个任务都应该实现 fang::AsyncRunnable 特性，该特性被 fang 用于执行它。

请注意不要使用相同名称的两个 AsyncRunnable 特性实现，因为这将在 typetag 包中导致失败。

use fang::AsyncRunnable;
use fang::asynk::async_queue::AsyncQueueable;
use fang::serde::{Deserialize, Serialize};
use fang::async_trait;

#[derive(Serialize, Deserialize)]
#[serde(crate = "fang::serde")]
struct AsyncTask {
    pub number: u16,
}

#[typetag::serde]
#[async_trait]
impl AsyncRunnable for AsyncTask {
    async fn run(&self, _queueable: &mut dyn AsyncQueueable) -> Result<(), Error> {
        Ok(())
    }
    // this func is optional
    // Default task_type is common
    fn task_type(&self) -> String {
        "my-task-type".to_string()
    }


    // If `uniq` is set to true and the task is already in the storage, it won't be inserted again
    // The existing record will be returned for for any insertions operaiton
    fn uniq(&self) -> bool {
        true
    }

    // This will be useful if you would like to schedule tasks.
    // default value is None (the task is not scheduled, it's just executed as soon as it's inserted)
    fn cron(&self) -> Option<Scheduled> {
        let expression = "0/20 * * * Aug-Sep * 2022/1";
        Some(Scheduled::CronPattern(expression.to_string()))
    }

    // the maximum number of retries. Set it to 0 to make it not retriable
    // the default value is 20
    fn max_retries(&self) -> i32 {
        20
    }

    // backoff mode for retries
    fn backoff(&self, attempt: u32) -> u32 {
        u32::pow(2, attempt)
    }
}

在这两个模块中，任务可以安排一次性执行。使用 Scheduled::ScheduleOnce 枚举变体。

日期和时间以及 cron 模式以 UTC 时区进行解释。因此，您应该引入偏移量来安排不同的时区。

示例

如果您的时区是 UTC + 2，并且您想在 11:00 安排

let expression = "0 0 9 * * * *";

入队任务

阻塞功能

要入队任务，请使用 Queue::enqueue_task

use fang::Queue;

// create a r2d2 pool

// create a fang queue

let queue = Queue::builder().connection_pool(pool).build();

let task_inserted = queue.insert_task(&MyTask::new(1)).unwrap();

异步功能

要入队任务，请使用 AsyncQueueable::insert_task。

对于 Postgres 后端

use fang::asynk::async_queue::AsyncQueue;
use fang::AsyncRunnable;

// Create an AsyncQueue
let max_pool_size: u32 = 2;

let mut queue = AsyncQueue::builder()
    // Postgres database url
    .uri("postgres://postgres:postgres@localhost/fang")
    // Max number of connections that are allowed
    .max_pool_size(max_pool_size)
    .build();

// Always connect first in order to perform any operation
queue.connect().await.unwrap();

加密始终与 rustls 包一起使用。我们计划在未来添加禁用它的方法。

// AsyncTask from the first example
let task = AsyncTask { 8 };
let task_returned = queue
    .insert_task(&task as &dyn AsyncRunnable)
    .await
    .unwrap();

启动工作者

阻塞功能

每个工作者都在一个单独的线程中运行。在发生 panic 的情况下，它们总是被重新启动。

使用 WorkerPool 来启动工作者。使用 WorkerPool::builder 来创建您的工作者池并运行任务。

use fang::WorkerPool;
use fang::Queue;

// create a Queue

let mut worker_pool = WorkerPool::<Queue>::builder()
    .queue(queue)
    .number_of_workers(3_u32)
    // if you want to run tasks of the specific kind
    .task_type("my_task_type")
    .build();

worker_pool.start();

异步功能

每个工作者都在一个单独的 tokio 任务中运行。在发生 panic 的情况下，它们总是被重新启动。使用 AsyncWorkerPool 来启动工作者。

use fang::asynk::async_worker_pool::AsyncWorkerPool;

// Need to create a queue
// Also insert some tasks

let mut pool: AsyncWorkerPool<AsyncQueue> = AsyncWorkerPool::builder()
        .number_of_workers(max_pool_size)
        .queue(queue.clone())
        // if you want to run tasks of the specific kind
        .task_type("my_task_type")
        .build();

pool.start().await;

查看

• 简单工作者示例 - 简单工作者示例
• 简单 Cron 工作者示例 - 简单工作者示例
• 简单 Async 工作者示例 - 简单 Async 工作者示例
• 简单 Cron Async 工作者示例 - 简单 Async 工作者示例
• El Monitorro - Telegram 信息订阅器。它使用 Fang 的阻塞模块来同步信息和向用户发送更新。
• weather_bot_rust - 一个提供天气信息的机器人。它使用 Fang 的 asynk 模块来处理来自 Telegram 用户的更新并安排天气信息。

配置

阻塞功能

只需使用 TypeBuilder 为 WorkerPool。

Asynk 功能

只需使用 TypeBuilder 为 AsyncWorkerPool。

配置工作者的类型

配置保留模式

默认情况下，所有成功完成的任务都会从数据库中删除，失败的不会。

您可以使用三种保留模式

pub enum RetentionMode {
    KeepAll,        // doesn't remove tasks
    RemoveAll,      // removes all tasks
    RemoveFinished, // default value
}

使用模块中的 TypeBuilder 设置保留模式。

配置睡眠值

阻塞功能

您可以使用 SleepParams 来配置睡眠值

pub struct SleepParams {
    pub sleep_period: Duration,     // default value is 5 seconds
    pub max_sleep_period: Duration, // default value is 15 seconds
    pub min_sleep_period: Duration, // default value is 5 seconds
    pub sleep_step: Duration,       // default value is 5 seconds
}

如果没有任务在数据库中，工作者会睡眠 sleep_period，并且每次这个值增加 sleep_step，直到它达到 max_sleep_period。 min_sleep_period 是 sleep_period 的初始值。所有值都是以秒为单位。

使用 set_sleep_params 来设置它

let sleep_params = SleepParams {
    sleep_period: Duration::from_secs(2),
    max_sleep_period: Duration::from_secs(6),
    min_sleep_period: Duration::from_secs(2),
    sleep_step: Duration::from_secs(1),
};

使用模块中的 TypeBuilder 设置睡眠参数。

贡献

1. 分支它！
2. 创建您的功能分支（git checkout -b my-new-feature）
3. 提交您的更改（git commit -am 'Add some feature'）
4. 将分支推送到（git push origin my-new-feature）
5. 创建一个新的拉取请求

在本地运行测试

• 安装 diesel_cli。

cargo install diesel_cli --no-default-features --features "postgres sqlite mysql"

• 在您的机器上安装 docker。

• 在您的机器上安装 SQLite 3。

• 为测试设置数据库。

make -j db

• 运行测试。 make db 不需要在每个测试周期之间运行。

make -j tests

• 运行脏/长测试。

make -j ignored

• 关闭数据库。

make -j stop

以上示例中的 -j 标志为 make 启用了并行性，不是必需的，但强烈推荐。

作者

• Ayrat Badykov (@ayrat555)

• Pepe Márquez (@pxp9)