Redis 任务队列

简单的 redis 任务队列

文档

https://docs.rs/rjq/

入队任务

extern crate rjq;

use std::time::Duration;
use std::thread::sleep;
use rjq::Queue;

fn main() {
    let queue = Queue::new("redis://127.0.0.1/", "rjq");
    let mut uuids = Vec::new();

    for _ in 0..10 {
        sleep(Duration::from_millis(100));
        uuids.push(queue.enqueue(vec![], 30).unwrap());
    }

    sleep(Duration::from_millis(10000));

    for uuid in uuids.iter() {
        let status = queue.status(uuid).unwrap();
        let result = queue.result(uuid).unwrap().unwrap();
        println!("{} {:?} {}", uuid, status, result);
    }
}

队列工作者

extern crate rjq;

use std::time::Duration;
use std::thread::sleep;
use std::error::Error;
use rjq::Queue;

fn main() {
    fn process(uuid: String, _: Vec<String>) -> Result<String, Box<Error>> {
        sleep(Duration::from_millis(1000));
        println!("{}", uuid);
        Ok(format!("hi from {}", uuid))
    }

    let queue = Queue::new("redis://127.0.0.1/", "rjq");
    queue.work(process, Some(1), Some(5), Some(10), Some(30), Some(false), None).unwrap();
}

任务状态

队列中 - 任务已入队等待进一步处理

运行中 - 任务正在由工作者运行

丢失 - 任务未及时完成

完成 - 任务已成功完成

失败 - 任务由于某些错误而失败

队列方法

初始化队列

fn new(url: &str, name: &str) -> Queue;

url - redis URL

name - 队列名称

返回 队列

删除队列任务

fn drop(&self) -> Result<(), Box<Error>>;

入队任务

fn enqueue(&self, args: Vec<String>, expire: usize) -> Result<String, Box<Error>>;

args - 任务参数

expire - 如果工作者在这段时间内没有启动任务，它将过期

返回任务的 UUID

获取任务状态

fn status(&self, uuid: &str) -> Result<Status, Box<Error>>;

uuid - 任务唯一标识符

返回任务的 状态

在队列上工作

fn work<F: Fn(String, Vec<String>) -> Result<String, Box<Error>> + Send + Sync + 'static>
    (&self,
     fun: F,
     wait: Option<usize>,
     timeout: Option<usize>,
     freq: Option<usize>,
     expire: Option<usize>,
     fall: Option<bool>,
     infinite: Option<bool>)
     -> Result<(), Box<Error>>;

fun - 工作者函数

wait - 等待下一个任务弹出直到这个时间

timeout - 工作者函数应该在超时（秒）内完成

freq - 任务状态检查频率（每秒次数）

expire - 任务结果将在这个时间内过期（秒）

fall - 如果任务丢失，则引发恐慌以终止进程

infinite - 无限次连续处理任务，否则只处理一个任务

获取任务结果

fn result(&self, uuid: &str) -> Result<Option<String>, Box<Error>>;

uuid - 任务唯一标识符

返回任务的 结果

运行测试

cargo test