Rust异步任务队列库Apalis的使用，高效处理后台任务和定时作业

安装

在项目目录中运行以下Cargo命令：

cargo add apalis

或在Cargo.toml中添加：

apalis = "0.7.2"

基本使用示例

以下是一个使用Apalis处理异步任务的完整示例：

use apalis::{prelude::*, redis::RedisStorage};
use serde::{Deserialize, Serialize};

// 定义一个任务
#[derive(Debug, Serialize, Deserialize)]
struct Email {
    to: String,
    message: String,
}

impl Job for Email {
    const NAME: &'static str = "apalis::Email";
}

// 任务处理器
async fn send_email(job: Email, _ctx: JobContext) -> Result<(), JobError> {
    println!("Sending email to {}: {}", job.to, job.message);
    Ok(())
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    // 创建Redis存储
    let storage = RedisStorage::connect("redis://127.0.0.1/").await.unwrap();
    
    // 创建工作者
    let worker = Worker::new("email-worker")
        .with_storage(storage.clone())
        .with_job::<Email>(send_email);
    
    // 启动工作者
    let monitor = Monitor::new()
        .register(worker)
        .run()
        .await;
    
    // 添加任务到队列
    storage
        .push(Email {
            to: "user@example.com".to_string(),
            message: "Hello from Apalis!".to_string(),
        })
        .await
        .unwrap();
    
    monitor.await
}

定时任务示例

Apalis还支持定时任务调度：

use apalis::{prelude::*, cron::CronStream, redis::RedisStorage};
use std::time::Duration;
use async_trait::async_trait;

#[derive(Debug, Clone)]
struct ScheduledJob;

#[async_trait]
impl Job for ScheduledJob {
    const NAME: &'static str = "apalis::ScheduledJob";
    
    async fn run(&self, _ctx: JobContext) -> Result<(), JobError> {
        println!("Running scheduled job at {}", chrono::Local::now());
        Ok(())
    }
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let storage = RedisStorage::connect("redis://127.0.0.1/").await.unwrap();
    
    // 创建定时任务流 (每5秒执行一次)
    let cron = CronStream::new("@every 5s")
        .duration(Duration::from_secs(5))
        .with_job(ScheduledJob);
    
    // 创建工作者
    let worker = Worker::new("cron-worker")
        .with_storage(storage.clone())
        .with_cron(cron);
    
    // 启动监控
    Monitor::new()
        .register(worker)
        .run()
        .await
        .await
}

完整示例：使用SQLite存储的任务队列

下面是一个使用SQLite作为存储后端的完整示例：

use apalis::{prelude::*, sqlite::SqliteStorage};
use serde::{Deserialize, Serialize};
use std::path::Path;

// 定义日志清理任务
#[derive(Debug, Serialize, Deserialize)]
struct CleanLogsJob {
    path: String,
    days_to_keep: u32,
}

impl Job for CleanLogsJob {
    const NAME: &'static str = "apalis::CleanLogsJob";
}

// 任务处理器
async fn clean_logs(job: CleanLogsJob, _ctx: JobContext) -> Result<(), JobError> {
    println!("Cleaning logs at path: {}, keeping last {} days", 
        job.path, job.days_to_keep);
    // 这里添加实际的日志清理逻辑
    Ok(())
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    // 创建SQLite存储
    let storage = SqliteStorage::new("sqlite::memory:").await.unwrap();
    
    // 初始化数据库表
    storage.setup().await.unwrap();
    
    // 创建工作者
    let worker = Worker::new("log-cleaner-worker")
        .with_storage(storage.clone())
        .with_job::<CleanLogsJob>(clean_logs);
    
    // 启动工作者
    let monitor = Monitor::new()
        .register(worker)
        .run()
        .await;
    
    // 添加多个任务到队列
    for i in 1..=5 {
        storage
            .push(CleanLogsJob {
                path: format!("/var/logs/service-{}", i),
                days_to_keep: 30,
            })
            .await
            .unwrap();
    }
    
    monitor.await
}

特性

1. 多种存储后端支持：Redis、SQLite、PostgreSQL等
2. 定时任务：通过Cron表达式调度任务
3. 任务重试：自动处理失败任务
4. 优先级队列：支持不同优先级的任务
5. 监控：提供任务执行监控接口

Apalis是一个功能强大的异步任务队列库，特别适合需要处理大量后台任务或定时作业的Rust应用程序。