Introduction

Installation

Add BullMQ to your project via Cargo:

cargo add bullmq-official --rename bullmq

Or add it to your Cargo.toml:

[dependencies]
bullmq = { version = "1.2", package = "bullmq-official" }

The crate is published as bullmq-official (the bullmq, bullmq-rust and bullmq-rs names on crates.io are already taken by unrelated third-party crates), but it is imported in code as bullmq — e.g. use bullmq::{Queue, Worker};.

BullMQ for Rust requires:

• Rust 1.85+
• Tokio runtime
• Redis 6.2+

Get Started

BullMQ uses Tokio for async processing. All operations are non-blocking and designed for high-throughput concurrent workloads.

Adding Jobs to a Queue

use bullmq::{Queue, QueueOptions};

#[tokio::main]
async fn main() -> bullmq::Result<()> {
    let queue = Queue::new("my-queue", QueueOptions::default()).await?;

    // Add a job with JSON data
    queue.add("my-job", serde_json::json!({
        "foo": "bar"
    }), None).await?;

    Ok(())
}

Processing Jobs with a Worker

use bullmq::{Worker, WorkerOptions, Job};
use bullmq::worker::{ProcessorFn, CancellationToken};
use std::sync::Arc;

#[tokio::main]
async fn main() -> bullmq::Result<()> {
    let processor: ProcessorFn = Arc::new(|job: Job, _token: CancellationToken| {
        Box::pin(async move {
            println!("Processing job: {} - {}", job.id(), job.name());
            // Access job data
            let data = job.data();
            println!("Data: {}", data);
            // Return a result value
            Ok(serde_json::json!({"processed": true}))
        })
    });

    let worker = Worker::new("my-queue", processor, WorkerOptions::default()).await?;

    // Worker processes jobs automatically in the background.
    // Wait for a signal or condition to shut down.
    tokio::signal::ctrl_c().await.unwrap();

    // Graceful shutdown with 5-second timeout
    worker.close(5000).await?;
    Ok(())
}

Listening to Worker Events

use bullmq::{Worker, WorkerOptions, Job};
use bullmq::worker::{ProcessorFn, CancellationToken};
use std::sync::Arc;

#[tokio::main]
async fn main() -> bullmq::Result<()> {
    let processor: ProcessorFn = Arc::new(|job: Job, _token: CancellationToken| {
        Box::pin(async move {
            Ok(serde_json::json!(null))
        })
    });

    let worker = Worker::new("my-queue", processor, WorkerOptions::default()).await?;

    // Consume events from the worker
    while let Some(event) = worker.next_event().await {
        match event {
            bullmq::worker::WorkerEvent::Completed { job_id, result } => {
                println!("Job {} completed with: {}", job_id, result);
            }
            bullmq::worker::WorkerEvent::Failed { job_id, error } => {
                println!("Job {} failed: {}", job_id, error);
            }
            bullmq::worker::WorkerEvent::Active { job_id } => {
                println!("Job {} started processing", job_id);
            }
            _ => {}
        }
    }

    Ok(())
}

Concurrency

Configure how many jobs are processed simultaneously:

let worker = Worker::new("my-queue", processor, WorkerOptions {
    concurrency: 10,
    ..Default::default()
}).await?;

The concurrency can be changed dynamically at runtime:

worker.set_concurrency(20);

Progress Tracking

Report progress from inside the processor:

let processor: ProcessorFn = Arc::new(|mut job: Job, _token: CancellationToken| {
    Box::pin(async move {
        for i in 0..100 {
            // Do work...
            job.update_progress(bullmq::JobProgress::Number(i as f64)).await?;
        }
        Ok(serde_json::json!(null))
    })
});

Job Retries with Backoff

use bullmq::{Queue, QueueOptions, JobOptions};
use bullmq::types::BackoffStrategy;

let queue = Queue::new("my-queue", QueueOptions::default()).await?;

queue.add("flaky-job", serde_json::json!({}), Some(JobOptions {
    attempts: Some(5),
    backoff: Some(BackoffStrategy::Exponential(1000)), // 1s, 2s, 4s, 8s, 16s
    ..Default::default()
})).await?;

Connection Configuration

use bullmq::{QueueOptions, WorkerOptions};
use bullmq::options::RedisConnectionOptions;

let conn = RedisConnectionOptions {
    url: "redis://user:password@redis.example.com:6380".to_string(),
    ..Default::default()
};

let queue = Queue::new("my-queue", QueueOptions {
    connection: conn.clone(),
    ..Default::default()
}).await?;

let worker = Worker::new("my-queue", processor, WorkerOptions {
    connection: conn,
    ..Default::default()
}).await?;

Instead of a URL you can use typed connection fields. When host is set, the URL is built from these fields (use tls: true for a rediss:// TLS connection):

use bullmq::options::RedisConnectionOptions;

let conn = RedisConnectionOptions {
    host: Some("redis.example.com".to_string()),
    port: Some(6380),
    username: Some("user".to_string()),
    password: Some("password".to_string()),
    db: Some(0),
    tls: true,
    ..Default::default()
};

Custom TLS certificates

To connect over TLS with a custom root CA (for example a self-signed certificate) or a client certificate and key for mutual TLS (mTLS), provide a tls_certs value. All certificates and keys must be in PEM format. Setting tls_certs implies a TLS (rediss://) connection, so tls: true is not required:

use bullmq::options::{RedisConnectionOptions, TlsCerts};

let conn = RedisConnectionOptions {
    host: Some("redis.example.com".to_string()),
    port: Some(6380),
    tls_certs: Some(TlsCerts {
        // Custom CA certificate (omit to use the default WebPKI root store).
        root_cert: Some(std::fs::read("ca.pem")?),
        // Client certificate and key for mutual TLS (mTLS).
        client_cert: Some(std::fs::read("client-cert.pem")?),
        client_key: Some(std::fs::read("client-key.pem")?),
    }),
    ..Default::default()
};

Key Differences from Node.js

Aspect	Node.js	Rust
Runtime	Event loop (single-threaded)	Tokio (multi-threaded async)
Processor	async function or sandboxed file	Arc<dyn Fn(Job, CancellationToken) -> Pin<Box<...>>>
Events	EventEmitter pattern	mpsc::UnboundedReceiver<WorkerEvent>
Error handling	Exceptions	Result<T, Error> types
Cancellation	AbortSignal	CancellationToken
Concurrency	Cooperative (single core)	True parallelism across all CPU cores

Compatibility

The Rust implementation uses the same Lua scripts and Redis data structures as the Node.js and Python versions. This means:

• Jobs added by Node.js workers can be processed by Rust workers (and vice versa)
• Queue state is fully shared across all language implementations
• You can mix languages in a single deployment