Hello, in this article , we are going to implement a message queue in node.js using BullMQ library which is built on the top of redis.
We will implement two message queues. One for adding refund task for a particular order. On successful completion of refund task, we will initiate notification task to notify user about refund completion. For notification task, we will use another queue.
Step 1: Setting up Project
Make a new folder "messaging_queue" and initialize it with npm init and add dependencies.
mkdir messaging_queue cd messaging_queue npm init npm i express bullmq -D Step 2: Implementation of Queue
Firstly, create a refundQueue.js file to write the code for implementing a refundQueue and a function to add refund tasks to refundQueue.
const { Queue } =require("bullmq") const refundQueue = new Queue("refund-queue",{ connection:{ host:"127.0.0.1", port:"6379" } }) async function addRefundTask(order){ const response = await refundQueue.add(`refund-to-${order.id}`,{ amount:order.amount, id:order.id, user_id:order.user_id }) console.log("Job added to refundQueue with id: ",response.id) } exports.addRefundTask=addRefundTask Now, create a notificationQueue.js file to write the code for implementing a notificationQueue and a function to add notification tasks to notificationQueue.
const { Queue } =require("bullmq") const notificationQueue = new Queue("notification-queue",{ connection:{ host:"127.0.0.1", port:"6379" } }) async function addNotificationTask(user_id,order_id){ const response = await notificationQueue.add(`notification-to-${user_id}`,{ order_id:order_id, user_id:user_id }) console.log("Job added to notificationQueue with id: ",response.id) } exports.addNotificationTask=addNotificationTask Step 3: Refund and Notification Process
Now, create two files refundProcess.js and notificationProcess.js to write the code for implementation of refund and notification process.
refundProcess.js
const refundComplete=()=>new Promise((res,rej)=>setTimeout(()=>res(),4*1000)); async function refundProcess(id,amount,user_id){ console.log(`Refund for order ${id} has started`); console.log(`Refund Amount: ${amount}`) console.log(`User ID: ${user_id}`) await refundComplete() console.log("Refund Completed Successfully!") } exports.refundProcess=refundProcess notificationProcess.js
const notificationComplete=()=>new Promise((res,rej)=>setTimeout(()=>res(),2*1000)); async function notificationProcess(user_id,order_id){ console.log(`Notify user ${user_id} about refund for order ${order_id}`) await notificationComplete() console.log("Notification Sent Successfully!") } exports.notificationProcess=notificationProcess Step 4: Implementation of Worker
Now, create a new file worker.js to implement refundWorker and notificationWorker.
const { Worker } =require("bullmq") const { refundProcess } =require("./refundProcess") const { notificationProcess } =require("./notificationProcess") const refundWorker = new Worker("refund-queue", async(job)=>{ console.log(`Refund Job ${job.id} started`) await refundProcess(job.data.id,job.data.amount,job.data.user_id) }) const notificationWorker = new Worker("notification-queue", async(job)=>{ console.log(`Notification Job ${job.id} started`) await notificationProcess(job.data.user_id,job.data.order_id) }) exports.refundWorker= refundWorker exports.notificationWorker=notificationWorker Step 5: Express Server
Now, in index.js file , write code for implementation of express server.
const express=require('express') const { addRefundTask } = require("./refundQueue") const { refundWorker } = require("./worker") const { addNotificationTask } = require('./notificationQueue') const { notificationWorker }=require('./worker') async function init(){ const app=express() const PORT = 8000; const order1={ id:"order1", amount:4000, user_id:"user1" } const order2={ id:"order2", amount:10000, user_id:"user2" } app.listen(PORT,()=>{ console.log("Server running at port 8000") }) await addRefundTask(order1); await addRefundTask(order2) refundWorker.on('completed', async(job) => { console.log(`Refund Job ${job.id} has completed!`); await addNotificationTask(job.data.user_id,job.data.id); }); refundWorker.on('failed', (job, err) => { console.log(`Refund Job ${job.id} has failed with ${err.message}`); }); notificationWorker.on('completed', (job) => { console.log(`Notification Job ${job.id} has completed!`); }); notificationWorker.on('failed', (job, err) => { console.log(`Notification Job ${job.id} has failed with ${err.message}`); }); } init() Here, we will add two orders to refundQueue.
We are using two event listeners 'completed' and 'failed' for both refundWorker and notificationWorker. On successful completion of refund task, a notification task is added to notificationQueue.
Step 6: Docker setup
We need a redis server running in local computer to run the code of BullMQ. So, we will use docker for that. Ensure docker is installed in your system. And, create a docker-compose.yml file.
version : '3.4' services: redis: container_name : redis-server image : redis ports: - 6379:6379 stdin_open : true Now, start the redis container using following command
docker compose up -d Now, we can run our express server
node index.js Here, is the output
Thank you for reading the article !
Top comments (1)
As is github code Compile/Run Suggestions:
As is github code Run Results:
BullMQ: DEPRECATION WARNING! Optional instantiation of Queue, Worker and QueueEvents without providing explicitly a connection or connection options is deprecated. This behaviour will be removed in the next major release