Queue

Main class for adding jobs to a queue and managing queue operations

Overview

The Queue class provides methods to add jobs to a queue and perform high-level queue administration such as pausing or deleting queues.

Constructor

new Queue(name: string, opts?: QueueOptions)

name · string · required

The name of the queue

opts · QueueOptions

Configuration options for the queue

Example

import { Queue } from 'bullmq';

const queue = new Queue('myQueue', {
  connection: {
    host: 'localhost',
    port: 6379,
  },
});

Methods

add

Adds a new job to the queue.

add(name: string, data: any, opts?: JobsOptions): Promise<Job>

name · string · required

Name of the job

data · any · required

Arbitrary data to append to the job

opts · JobsOptions

Job options that affect how the job is processed

job · Job

The created job instance

addBulk

Adds an array of jobs to the queue atomically.

addBulk(jobs: Array<{ name: string; data: any; opts?: JobsOptions }>): Promise<Job[]>

jobs · Array · required

Array of job objects with name, data, and opts properties

jobs · Job[]

Array of created job instances

pause

Pauses the processing of this queue globally.

pause(): Promise<void>

resume

Resumes the processing of this queue globally.

resume(): Promise<void>

isPaused

Returns true if the queue is currently paused.

isPaused(): Promise<boolean>

getJob

Fetches a job by its ID.

getJob(jobId: string): Promise<Job | undefined>

jobId · string · required

The job ID to fetch

getJobs

Returns jobs on the given statuses.

getJobs(types?: JobType[], start?: number, end?: number, asc?: boolean): Promise<Job[]>

types · JobType[]

Job statuses to retrieve (e.g., ‘waiting’, ‘active’, ‘completed’)

start · number · default: "0"

Zero-based index from where to start returning jobs

end · number · default: "-1"

Zero-based index where to stop returning jobs

asc · boolean · default: "false"

If true, jobs will be returned in ascending order

getJobCounts

Returns the job counts for each type specified.

getJobCounts(...types: JobType[]): Promise<{ [index: string]: number }>

clean

Cleans jobs from a queue within a certain grace period.

clean(grace: number, limit: number, type?: string): Promise<string[]>

grace · number · required

The grace period in milliseconds

limit · number · required

Max number of jobs to clean

type · string · default: "'completed'"

The type of job to clean: ‘completed’, ‘wait’, ‘active’, ‘paused’, ‘delayed’, or ‘failed’

drain

Drains the queue, removing all jobs that are waiting or delayed.

drain(delayed?: boolean): Promise<void>

delayed · boolean · default: "false"

If true, also clean delayed jobs

obliterate

Completely destroys the queue and all of its contents irreversibly.

obliterate(opts?: { force?: boolean; count?: number }): Promise<void>

opts.force · boolean · default: "false"

Force obliteration even with active jobs in the queue

opts.count · number · default: "1000"

Maximum number of deleted keys per iteration

remove

Removes a job from the queue.

remove(jobId: string, opts?: { removeChildren?: boolean }): Promise<number>

jobId · string · required

The ID of the job to remove

opts.removeChildren · boolean · default: "true"

Whether to remove child jobs

setGlobalConcurrency

Enables and sets global concurrency value.

setGlobalConcurrency(concurrency: number): Promise<void>

concurrency · number · required

Maximum number of simultaneous jobs that workers can handle

removeGlobalConcurrency

Removes the global concurrency limit.

removeGlobalConcurrency(): Promise<void>

setGlobalRateLimit

Enables and sets rate limit.

setGlobalRateLimit(max: number, duration: number): Promise<void>

max · number · required

Max number of jobs to process in the time period

duration · number · required

Time in milliseconds for the rate limit period

removeGlobalRateLimit

Removes global rate limit.

removeGlobalRateLimit(): Promise<void>

upsertJobScheduler

Upserts a job scheduler for creating jobs at intervals.

upsertJobScheduler(
  jobSchedulerId: string,
  repeatOpts: RepeatOptions,
  jobTemplate?: { name?: string; data?: any; opts?: JobsOptions }
): Promise<Job>

jobSchedulerId · string · required

Unique identifier for the job scheduler

repeatOpts · RepeatOptions · required

Repeat configuration options

jobTemplate · object

Template for jobs created by this scheduler

getJobSchedulers

Get all job schedulers.

getJobSchedulers(start?: number, end?: number, asc?: boolean): Promise<JobSchedulerJson[]>

removeJobScheduler

Removes a job scheduler.

removeJobScheduler(jobSchedulerId: string): Promise<boolean>

close

Closes the queue instance.

close(): Promise<void>

Events

The Queue class extends EventEmitter and emits the following events:

• waiting - A job has been added to the queue
• progress - A job’s progress has been updated
• completed - A job has completed successfully
• failed - A job has failed
• paused - The queue has been paused
• resumed - The queue has been resumed
• removed - A job has been removed
• cleaned - Jobs have been cleaned from the queue
• error - An error occurred