BullMQ Documentation

BullMQ is a Node.js library that provides a fast and robust queue system built on top of Redis. It is designed for modern micro-services architectures and focuses on high performance, horizontal scalability, and consistency.

Key architectural goals include:

• Exactly once queue semantics: Attempts to deliver every message exactly one time (delivering at least once in worst-case scenarios).
• Horizontal Scalability: Easily add more workers to process jobs in parallel.
• High Performance: Maximizes Redis throughput using efficient .lua scripts and pipelining.
• Polling-free design: Minimizes CPU usage.

BullMQ Pro offers advanced capabilities for enterprise use cases:

• Observables: Support for job cancellation via observables.
• Groups: Advanced grouping features including group-level rate limiting, concurrency, and pausing.
• Batches: Efficiently processing multiple jobs at once.
• NestJs Integration: Specialized support for NestJs producers and event listeners.

When using a FlowProducer to add jobs, jobs can have dependencies on other jobs (children). This introduces an additional state:

• waiting-children: A job enters this state when it has children that must complete before the parent can be processed. The parent job is not processed directly; instead, it is placed in the wait, delayed, or prioritized set automatically as soon as the last child job is marked as completed.

BullMQ is built around four primary classes that work together to manage job processing:

• Queue: Used to add jobs to the queue and perform basic manipulations like pausing, cleaning, or retrieving queue data.
• Worker: Instances responsible for consuming and processing jobs from a queue. Workers can run in the same Node.js process, in separate processes, or on different machines.
• QueueEvents: Used to listen to events happening within the queue.
• FlowProducer: Used to create complex job flows (parent/child relationships).

When you add a job to a Queue using the add method, it moves through several states. Understanding these states is critical for managing job flow and processing logic.

Standard Queue Job States

• wait: The initial waiting list where jobs reside before being picked up for processing.
• prioritized: Jobs with a specific priority level. Higher priority jobs are processed before lower priority ones.
• delayed: Jobs waiting for a specific timeout. Once the timeout expires, they are moved to the wait list or prioritized set to be processed as soon as a worker is idle.
• active: Jobs currently being executed by a worker (running inside the process function).
• completed: The terminal state for a successfully processed job.
• failed: The terminal state for a job that encountered an exception during processing.

To prevent blocking the main event loop or to isolate crashes, you can run processors in separate processes (sandboxed). This allows for better multi-core CPU utilization and prevents jobs from stalling the queue.

1. Create a separate file for your processor logic.
2. Pass the file path to the .process() method.

// processor.js
module.exports = function (job) {
  // Do some heavy work
  return Promise.resolve(result);
}

// In your main application:
// Single process:
queue.process('/path/to/my/processor.js');

// With concurrency:
queue.process(5, '/path/to/my/processor.js');

// With a named processor:
queue.process('my processor', 5, '/path/to/my/processor.js');

BullMQ uses ioredis by default. For production robustness, understand these key options:

• retryStrategy: Determines the reconnection function. BullMQ's default uses exponential backoff (1s to 20s).
• maxRetriesPerRequest: For Worker classes, this must be set to null to prevent Redis exceptions from breaking worker functionality. BullMQ sets this to null by default for Workers, but it can be overridden if you pass an existing instance.
• enableOfflineQueue: For Queue (producers), you should typically disable this so calls fail quickly during downtime. For Worker (consumers), you should leave it enabled so they wait for reconnection.

// Example of custom retryStrategy
retryStrategy: function (times: number) {
  return Math.max(Math.min(Math.exp(times), 20000), 1000);
}