Worker

Process jobs from a queue with concurrency support

Overview

The Worker class represents a worker that processes jobs from the queue. As soon as the class is instantiated and a connection to Redis is established, it will start processing jobs.

Constructor

new Worker(
  name: string,
  processor?: Processor | string | URL,
  opts?: WorkerOptions
)

name · string · required

The name of the queue to process

processor · Processor | string | URL

The processor function or path to processor file

opts · WorkerOptions

Configuration options for the worker

Example

import { Worker } from 'bullmq';

const worker = new Worker('myQueue', async (job) => {
  // Process the job
  console.log('Processing job', job.id, job.data);
  return { processed: true };
}, {
  connection: {
    host: 'localhost',
    port: 6379,
  },
  concurrency: 5,
});

Properties

id

readonly id: string

Unique identifier for this worker instance.

concurrency

concurrency: number

The maximum number of jobs this worker will process concurrently.

opts

readonly opts: WorkerOptions

The worker configuration options.

Methods

run

Starts processing jobs from the queue.

run(): Promise<void>

pause

Pauses the processing of jobs for this worker.

pause(doNotWaitActive?: boolean): Promise<void>

doNotWaitActive · boolean · default: "false"

If true, does not wait for active jobs to finish

resume

Resumes processing jobs after being paused.

resume(): void

isPaused

Checks if the worker is currently paused.

isPaused(): boolean

isRunning

Checks if the worker is currently running.

isRunning(): boolean

getNextJob

Manually fetches the next job from the queue.

getNextJob(token: string, opts?: { block?: boolean }): Promise<Job | undefined>

token · string · required

Worker token to assign to retrieved job

opts.block · boolean · default: "true"

Whether to block waiting for a job

cancelJob

Cancels a specific job currently being processed.

cancelJob(jobId: string, reason?: string): boolean

jobId · string · required

The ID of the job to cancel

reason · string

Optional reason for cancellation

cancelled · boolean

True if the job was found and cancelled

cancelAllJobs

Cancels all jobs currently being processed by this worker.

cancelAllJobs(reason?: string): void

reason · string

Optional reason for cancellation

close

Closes the worker and related Redis connections.

close(force?: boolean): Promise<void>

force · boolean · default: "false"

If true, does not wait for current jobs to be processed

startStalledCheckTimer

Manually starts the stalled job checker.

startStalledCheckTimer(): Promise<void>

Static Methods

RateLimitError

Creates a rate limit error to throw from a processor.

static RateLimitError(): Error

Events

The Worker class extends EventEmitter and emits the following events:

active

Emitted when a job enters the active state.

worker.on('active', (job: Job, prev: string) => {
  console.log(`Job ${job.id} is now active`);
});

completed

Emitted when a job has successfully completed.

worker.on('completed', (job: Job, result: any, prev: string) => {
  console.log(`Job ${job.id} completed with result:`, result);
});

failed

Emitted when a job has failed.

worker.on('failed', (job: Job | undefined, error: Error, prev: string) => {
  console.log(`Job ${job?.id} failed:`, error.message);
});

progress

Emitted when a job updates its progress.

worker.on('progress', (job: Job, progress: number | object) => {
  console.log(`Job ${job.id} progress:`, progress);
});

drained

Emitted when the queue has drained the waiting list.

worker.on('drained', () => {
  console.log('Queue is drained');
});

paused

Emitted when the worker is paused.

worker.on('paused', () => {
  console.log('Worker paused');
});

resumed

Emitted when the worker is resumed.

worker.on('resumed', () => {
  console.log('Worker resumed');
});

stalled

Emitted when a job has stalled.

worker.on('stalled', (jobId: string, prev: string) => {
  console.log(`Job ${jobId} stalled`);
});

error

Emitted when an error occurs.

worker.on('error', (error: Error) => {
  console.error('Worker error:', error);
});

closing

Emitted when the worker is closing.

worker.on('closing', (msg: string) => {
  console.log('Worker closing:', msg);
});

closed

Emitted when the worker has closed.

worker.on('closed', () => {
  console.log('Worker closed');
});