Consumer
- Node.js course
- Advanced
- Lesson 58/83
- 5 min
Learn job processing with a Bull consumer/process.
Consumer
The consumer is the side that takes messages from the queue and processes them. In Bull you define a consumer with queue.process((job) => { ... }). The callback is called for each message; if it returns a Promise, Bull waits for it to finish.
Example: kuyruk.process((job) => { console.log("Processing: " + job.data.action); return Promise.resolve(); });. When a message is taken from the queue, the callback is called with a job parameter. job.data is the message's data. If the callback returns a Promise, resolve means success, reject means failure. Failed jobs are retried. Multiple consumers can process the queue in parallel.
job.data and Promise
The consumer callback takes a job parameter. job.data is the message's data. The callback should return a Promise; resolve means success, reject means failure.
Example: kuyruk.process(async (job) => { const user = job.data.user; await sendMail(user); });. An async function automatically returns a Promise, so no explicit return is needed. If sendMail succeeds, the job is completed; if it throws, Bull marks the job as failed and applies the retry rule. job.attemptsMade tells how many times it has been tried. Since the consumer is asynchronous, multiple jobs can run in parallel; Bull limits parallelism with { concurrency: 5 }.
Practice Define a consumer
Use the kuyruk object below. Define a consumer with kuyruk.process(async (job) => { console.log("Job: " + job.data.action); }). Print "Consumer ready" with console.log.
Starter code
const Bull = require("bull");
const kuyruk = new Bull("job-queue");
// Define a consumer with kuyruk.process
// write here
// Print "Consumer ready"
// write hereSample solution
const Bull = require("bull");
const kuyruk = new Bull("job-queue");
kuyruk.process(async (job) => {
console.log("Job: " + job.data.action);
});
console.log("Consumer ready");Expected output
Consumer ready
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 What does queue.process do?
kuyruk.process((job) => { ... })- Adds a message to the queue
- Takes and processes messages Correct answer
- Closes Redis
- Creates a new queue
Explanation: queue.process takes and processes messages.
2 Complete the job.data access
const user = job.____.user;const user = job.data.user;Explanation: job.data accesses the message data.
3 Match each piece to its role
| Term | Meaning |
|---|---|
queue.process |
Defines the consumer |
job.data |
Message data |
Promise resolve |
Job succeeded |
Promise reject |
Job failed |