Crate asynq 

Asynq

Simple, reliable & efficient distributed task queue in Rust

Asynq 是一个分布式任务队列库，用于在 Redis 支持的异步工作者之间排队任务并处理任务。 Asynq is a distributed task queue library for queuing and processing tasks among async workers backed by Redis. 它旨在具有可扩展性，但易于上手。 It is designed to be scalable yet easy to use.

特性

Features

• 保证任务至少执行一次
  • Guarantees at-least-once task execution
• 任务调度功能
  • Task scheduling capability
• 失败任务的重试机制
  • Retry mechanism for failed tasks
• 在工作者崩溃时自动恢复任务
  • Automatic task recovery on worker crash
• 加权优先级队列
  • Weighted priority queues
• 严格优先级队列
  • Strict priority queues
• 低延迟添加任务，因为 Redis 写入速度快
  • Low-latency task addition due to fast Redis writes
• 使用唯一选项去重任务
  • Deduplication of tasks using unique option
• 支持每个任务的超时和截止时间
  • Supports timeout and deadline for each task
• 允许聚合任务组以批处理多个连续操作
  • Allows aggregation of task groups for batch processing
• 灵活的处理器接口，支持中间件
  • Flexible handler interface with middleware support
• 能够暂停队列以停止处理来自队列的任务
  • Ability to pause queues to stop processing tasks from a queue
• 周期性任务
  • Periodic tasks
• 支持 Redis Sentinels 实现高可用性
  • Supports Redis Sentinels for high availability
• 与 Prometheus 集成以收集和可视化队列指标
  • Integration with Prometheus for queue metrics collection and visualization
• Web UI 检查和远程控制队列和任务
  • Web UI for inspecting and remotely controlling queues and tasks

多后端支持

Multiple Backend Support

现在支持同时启用多个后端，默认使用 Redis 后端。 Now supports enabling multiple backends simultaneously, with Redis as the default.

• Redis: 始终可用作为默认后端
• Redis: Always available as the default backend
• PostgreSQL: 需要启用 postgres feature
• PostgreSQL: Requires postgres feature to be enabled
• WebSocket: 需要启用 websocket feature
• WebSocket: Requires websocket feature to be enabled

快速开始

Quick Start

使用 Redis 后端

Using Redis Backend

use asynq::{client::Client,task::Task};
use asynq::server::{Server,Handler};
use async_trait::async_trait;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 创建 Redis 配置
    // Create Redis configuration
    use asynq::backend::RedisConnectionType;
    let redis_config = asynq::backend::RedisConnectionType::single("redis://127.0.0.1:6379")?;
     
    // 创建客户端
    // Create client
    let client = Client::new(redis_config).await?;
     
    // 创建任务
    // Create task
    let task = Task::new("email:deliver", b"task payload").unwrap();
     
    // 将任务加入队列
    // Enqueue task
    client.enqueue(task).await?;
     
    Ok(())
}

使用 PostgresSQL 后端 (需要 postgres feature)

Using PostgresSQL Backend (requires postgres feature)

use asynq::{client::Client,task::Task};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 创建 PostgresSQL 客户端
    // Create PostgresSQL client
    let client = Client::new_with_postgres("postgres://user:pass@localhost/dbname").await?;
     
    // 创建任务
    // Create task
    let task = Task::new("email:deliver", b"task payload").unwrap();
     
    // 将任务加入队列
    // Enqueue task
    client.enqueue(task).await?;
     
    Ok(())
}

使用 WebSocket 后端连接 asynq-server (需要 websocket feature)

Using WebSocket Backend to connect to asynq-server (requires websocket feature)

use asynq::{client::Client, task::Task};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 连接到 asynq-server
    // Connect to asynq-server
    let client = Client::new_with_websocket("ws://127.0.0.1:8080/ws").await?;
     
    // 创建任务
    // Create task
    let task = Task::new("email:deliver", b"task payload").unwrap();
     
    // 将任务加入队列
    // Enqueue task
    client.enqueue(task).await?;
     
    Ok(())
}

Modules

backend

后端模块 Backend module

base

经纪人模块 Broker module

client

客户端模块 Client module

components

Lifecycle 模块 Lifecycle module

config

配置模块 Configuration module

error

错误处理模块 Error handling module

inspector

检查器模块 Inspector module

proto

Protocol related code. source: asynq.proto

scheduler

周期性任务调度器（Scheduler）

serve_mux

ServeMux - 任务路由多路复用器 ServeMux - Task routing multiplexer

server

服务器模块 Server module

task

任务模块 Task module