Rust异步任务队列库apalis-core的使用：高效处理后台作业与分布式任务调度

介绍

apalis-core是一个Rust编写的异步任务队列库，专注于高效处理后台作业和分布式任务调度。它提供了以下核心特性：

• 基于async/await的异步任务处理
• 支持多种存储后端（Redis、PostgreSQL、SQLite等）
• 任务重试机制
• 任务优先级支持
• 分布式任务调度能力
• 与Tokio运行时深度集成

安装

在Cargo.toml中添加依赖：

[dependencies]
apalis-core = "0.5"
tokio = { version = "1.0", features = ["full"] }

基本使用方法

1. 定义任务

use apalis_core::job::{Job, JobContext};

#[derive(Debug, Clone)]
struct EmailJob {
    to: String,
    subject: String,
    body: String,
}

#[async_trait::async_trait]
impl Job for EmailJob {
    type Output = ();

    async fn run(self, ctx: JobContext) -> Result<Self::Output, anyhow::Error> {
        println!("Sending email to {} with subject '{}'", self.to, self.subject);
        // 这里实际发送邮件...
        Ok(())
    }
}

2. 创建并运行Worker

use apalis_core::{builder::WorkerBuilder, storage::SqliteStorage};

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    // 使用SQLite作为存储后端
    let storage = SqliteStorage::connect("sqlite://jobs.db").await?;
    
    // 创建worker
    let worker = WorkerBuilder::new("email-worker")
        .with_storage(storage)
        .build::<EmailJob>();
    
    // 运行worker
    worker.run().await?;
    
    Ok(())
}

3. 提交任务

use apalis_core::storage::Storage;

async fn submit_email_job(storage: &SqliteStorage) -> Result<(), anyhow::Error> {
    let job = EmailJob {
        to: "user@example.com".to_string(),
        subject: "Welcome".to_string(),
        body: "Thank you for signing up!".to_string(),
    };
    
    storage.push(job).await?;
    Ok(())
}

高级功能

任务重试

#[async_trait::async_trait]
impl Job for EmailJob {
    type Output = ();
    
    // 设置最大重试次数
    fn max_retries(&self) -> i32 {
        3
    }
    
    // 设置重试间隔
    fn backoff(&self, attempt: u32) -> std::time::Duration {
        std::time::Duration::from_secs(60 * attempt as u64)
    }
    
    async fn run(self, ctx: JobContext) -> Result<Self::Output, anyhow::Error> {
        // 实现略...
    }
}

使用Redis作为后端

use apalis_core::storage::RedisStorage;

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    let redis_url = "redis://127.0.0.1/";
    let storage = RedisStorage::connect(redis_url).await?;
    
    let worker = WorkerBuilder::new("redis-worker")
        .with_storage(storage)
        .build::<EmailJob>();
    
    worker.run().await?;
    Ok(())
}

分布式任务调度

use apalis_core::scheduler::Scheduler;

async fn schedule_daily_report(storage极抱歉，您的内容似乎被截断了。我将基于您提供的完整内容，给出一个完整的示例demo：

```rust
//! 完整示例：使用apalis-core实现电子邮件任务队列

use apalis_core::{
    builder::WorkerBuilder,
    job::{Job, JobContext},
    scheduler::Scheduler,
    storage::{RedisStorage, SqliteStorage, Storage},
};
use async_trait::async_trait;
use chrono::Utc;
use std::time::Duration;
use tokio::time::sleep;

// 1. 定义任务类型
#[derive(Debug, Clone)]
struct EmailJob {
    to: String,
    subject: String,
    body: String,
}

// 2. 实现Job trait
#[async_trait]
impl Job for EmailJob {
    type Output = ();

    // 最大重试次数
    fn max_retries(&self) -> i32 {
        3
    }

    // 重试间隔
    fn backoff(&self, attempt: u32) -> Duration {
        Duration::from_secs(60 * attempt as u64)
    }

    async fn run(self, ctx: JobContext) -> Result<Self::Output, anyhow::Error> {
        println!(
            "[{}] Processing email to: {}, subject: {}",
            Utc::now(),
            self.to,
            self.subject
        );
        
        // 模拟邮件发送失败
        if self.to.contains("fail") {
            return Err(anyhow::anyhow!("Failed to send email to {}", self.to));
        }
        
        // 模拟邮件发送成功
        sleep(Duration::from_secs(1)).await;
        println!("Email sent successfully to {}", self.to);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    // 3. 初始化存储后端 (这里使用Redis)
    let redis_url = "redis://127.0.0.1/";
    let storage = RedisStorage::connect(redis_url).await?;

    // 4. 创建并启动worker
    let worker = WorkerBuilder::new("email-worker")
        .with_storage(storage.clone())
        .build::<EmailJob>();

    tokio::spawn(async move {
        worker.run().await.expect("Worker failed");
    });

    // 5. 添加定时任务
    let scheduler = Scheduler::new(storage.clone());
    scheduler
        .cron("0 * * * * *", || {
            // 每分钟执行一次
            EmailJob {
                to: "cron@example.com".to_string(),
                subject: "Scheduled Email".to_string(),
                body: "This is a scheduled email".to_string(),
            }
        })
        .await?;

    // 6. 提交即时任务
    storage
        .push(EmailJob {
            to: "user@example.com".to_string(),
            subject: "Welcome".to_string(),
            body: "Thank you for signing up!".to_string(),
        })
        .await?;

    // 7. 提交会失败的任务（测试重试机制）
    storage
        .push(EmailJob {
            to: "fail@example.com".to_string(),
            subject: "Test Failure".to_string(),
            body: "This should fail".to_string(),
        })
        .await?;

    // 保持程序运行
    loop {
        sleep(Duration::from_secs(10)).await;
    }
}

这个完整示例演示了：

1. 定义EmailJob任务类型
2. 实现Job trait并配置重试策略
3. 使用Redis作为存储后端
4. 创建并运行worker处理任务
5. 设置定时任务（每分钟执行）
6. 提交即时任务
7. 测试失败重试机制

要运行此示例，需要：

1. 安装并运行Redis服务器
2. 在Cargo.toml中添加依赖：

   [dependencies]
   apalis-core = "0.5"
   tokio = { version = "1.0", features = ["full"] }
   async-trait = "0.1"
   anyhow = "1.0"
   chrono = "0.4"