BullMQ
Search…
Guide
Patterns
Bull 3.x Migration
Workers
Workers are the actual instances that perform some job based on the jobs that are added in the queue. A worker is equivalent to a "message" receiver in a traditional message queue. The worker duty is to complete the job, if it succeeds the job will be moved to the "completed" status. If the worker throws an exception during its processing, the job will automatically be moved to the "failed" status.
A worker is instantiated with the Worker class, and the work itself will be performed in the process function. Process functions are meant to be asynchronous so either use the "async" keyword or return a promise.
1
import { Worker, Job } from 'bullmq';
2
​
3
const worker = new Worker(queueName, async (job: Job) => {
4
// Optionally report some progress
5
await job.updateProgress(42);
6
​
7
// Optionally sending an object as progress
8
await job.updateProgress({ foo: 'bar' });
9
​
10
// Do something with job
11
return 'some value';
12
});
Copied!
When a worker instance is created, it launches the processor immediately
In order to decide when your processor should start its execution, pass autorun as false as part of worker options:
1
import { Worker, Job } from 'bullmq';
2
​
3
const worker = new Worker(
4
queueName,
5
async (job: Job) => {
6
// Optionally report some progress
7
await job.updateProgress(42);
8
​
9
// Optionally sending an object as progress
10
await job.updateProgress({ foo: 'bar' });
11
​
12
// Do something with job
13
return 'some value';
14
},
15
{ autorun: false },
16
);
17
​
18
worker.run();
Copied!
Note that a processor can optionally return a value. This value can be retrieved either by getting the job and accessing the "returnvalue" property or by listening to the "completed" event:
1
worker.on('completed', (job: Job, returnvalue: any) => {
2
// Do something with the return value.
3
});
Copied!
Inside the worker process function it is also possible to emit progress events. Calling "job.progress" you can specify a number or an object if you have more complex needs. The "progress" event can be listened in the same way as the "completed" event:
1
worker.on('progress', (job: Job, progress: number | object) => {
2
// Do something with the return value.
3
});
Copied!
Finally, when the process fails with an exception it is possible to listen for the "failed" event too:
1
worker.on('failed', (job: Job, error: Error) => {
2
// Do something with the return value.
3
});
Copied!
It is also possible to listen to global events in order to get notifications of job completions, progress and failures:
1
import { QueueEvents } from 'bullmq';
2
​
3
const queueEvents = new QueueEvents('Paint');
4
​
5
queueEvents.on('completed', ({ jobId: string, returnvalue: any }) => {
6
// Called every time a job is completed in any worker.
7
});
8
​
9
queueEvents.on('failed', ({ jobId: string, failedReason: string }) => {
10
// jobId received a progress event
11
});
12
​
13
queueEvents.on('progress', ({jobId: string, data: number | object}) => {
14
// jobId received a progress event
15
});
Copied!
Finally, you should attach an error listener to your worker to avoid NodeJS raising an unhandled exception when an error occurs, something like this:
1
worker.on('error', err => {
2
// log the error
3
console.error(err);
4
});
Copied!
If the error handler is missing, your worker may stop processing jobs when an error is emitted!. More info here.
Typescript typings
It is also possible to specify the data types for the Job data and return value using generics:
1
const worker = new Worker<MyData, MyReturn>(queueName, async (job: Job) => {});
Copied!
Read more:
Last modified 1mo ago
Copy link