Queue producers in NestJS

Introduction

A queue producer sends messages or tasks to a queue so other parts of the app can process them later. It helps keep things organized and smooth.

When you want to handle tasks in the background without making users wait.

When you need to send data to another service or microservice asynchronously.

When you want to balance workload by distributing tasks over time.

When you want to retry failed tasks without blocking the main app.

When you want to decouple parts of your app for better maintenance.

Syntax

NestJS

import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Injectable()
export class MyProducerService {
  constructor(@InjectQueue('myQueue') private readonly myQueue: Queue) {}

  async sendMessage(data: any) {
    await this.myQueue.add(data);
  }
}

Use @InjectQueue('queueName') to get the queue instance.

Use queue.add(data) to send a message to the queue.

Examples

Sends an empty object to the queue. Useful to trigger a task without data.

NestJS

import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Injectable()
export class EmptyDataProducer {
  constructor(@InjectQueue('emptyQueue') private readonly emptyQueue: Queue) {}

  async sendEmpty() {
    await this.emptyQueue.add({});
  }
}

Sends a single task with details to the queue.

NestJS

import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Injectable()
export class SingleItemProducer {
  constructor(@InjectQueue('singleItemQueue') private readonly singleItemQueue: Queue) {}

  async sendSingleItem() {
    await this.singleItemQueue.add({ task: 'sendEmail', to: '[e-mail]' });
  }
}

Sends multiple different tasks to the same queue.

NestJS

import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Injectable()
export class MultipleItemsProducer {
  constructor(@InjectQueue('multiQueue') private readonly multiQueue: Queue) {}

  async sendMultiple() {
    await this.multiQueue.add({ task: 'generateReport', reportId: 123 });
    await this.multiQueue.add({ task: 'cleanup', olderThanDays: 30 });
  }
}

Sample Program

This NestJS module sets up a queue named 'tasks' connected to Redis. The QueueProducerService sends a task message when the app starts.

NestJS

import { Module, Injectable, OnModuleInit } from '@nestjs/common';
import { BullModule, InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Injectable()
export class QueueProducerService implements OnModuleInit {
  constructor(@InjectQueue('tasks') private readonly tasksQueue: Queue) {}

  async onModuleInit() {
    console.log('Sending initial task to queue...');
    await this.sendTask({ type: 'email', to: '[e-mail]', subject: 'Hello!' });
    console.log('Task sent.');
  }

  async sendTask(data: any) {
    await this.tasksQueue.add(data);
  }
}

@Module({
  imports: [
    BullModule.forRoot({
      redis: { host: 'localhost', port: 6379 },
    }),
    BullModule.registerQueue({ name: 'tasks' }),
  ],
  providers: [QueueProducerService],
})
export class AppModule {}

// To run this example, start a Redis server locally.
// Then start NestJS app. It will send a task to the 'tasks' queue on startup.

Important Notes

Adding a job to the queue is usually fast and non-blocking.

Make sure Redis is running and accessible for Bull queues to work.

Common mistake: forgetting to register the queue in the module imports.

Summary

Queue producers send tasks or messages to queues for later processing.

Use @InjectQueue to get the queue and queue.add() to send data.

This helps keep your app responsive and organized by handling work asynchronously.