Documentation Index
Fetch the complete documentation index at: https://mintlify.com/taskforcesh/bullmq/llms.txt
Use this file to discover all available pages before exploring further.
Job producers add jobs to queues in NestJS. Producers are typically application services (Nest providers) that inject queues to add jobs.
Basic Producer
Inject the queue into your service
Use the @InjectQueue() decorator to inject a queue by name:import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
@Injectable()
export class AudioService {
constructor(@InjectQueue('audio') private audioQueue: Queue) {}
}
The @InjectQueue() decorator identifies the queue by its name, as provided in registerQueue().
Add jobs to the queue
Call the queue’s add() method to add jobs:async convertAudio(file: string, format: string) {
const job = await this.audioQueue.add('convert', {
file,
format,
});
return job.id;
}
Complete Producer Example
import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
@Injectable()
export class AudioService {
constructor(@InjectQueue('audio') private audioQueue: Queue) {}
async convertAudio(file: string, format: string) {
const job = await this.audioQueue.add('convert', {
file,
format,
});
return {
jobId: job.id,
status: 'queued',
};
}
async convertWithOptions(file: string, format: string) {
const job = await this.audioQueue.add(
'convert',
{
file,
format,
},
{
// Job options
priority: 1,
delay: 5000, // Delay 5 seconds
attempts: 3,
backoff: {
type: 'exponential',
delay: 1000,
},
}
);
return job.id;
}
async getJobStatus(jobId: string) {
const job = await this.audioQueue.getJob(jobId);
if (!job) {
return null;
}
return {
id: job.id,
state: await job.getState(),
progress: job.progress,
data: job.data,
};
}
}
Job Options
Customize job behavior with various options:
Priority
Delay
Attempts & Backoff
Remove on Complete
await this.audioQueue.add('convert', data, {
priority: 1, // Lower number = higher priority
});
await this.audioQueue.add('convert', data, {
delay: 5000, // Delay 5 seconds
});
await this.audioQueue.add('convert', data, {
attempts: 3,
backoff: {
type: 'exponential',
delay: 1000,
},
});
await this.audioQueue.add('convert', data, {
removeOnComplete: true,
removeOnFail: 100, // Keep last 100 failed jobs
});
Flow Producers
For complex workflows with parent-child job relationships, use FlowProducer:
Register the FlowProducer
import { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
@Module({
imports: [
BullModule.registerFlowProducer({
name: 'videoflow',
}),
],
})
export class VideoModule {}
Inject the FlowProducer
import { Injectable } from '@nestjs/common';
import { InjectFlowProducer } from '@nestjs/bullmq';
import { FlowProducer } from 'bullmq';
@Injectable()
export class VideoService {
constructor(
@InjectFlowProducer('videoflow') private flowProducer: FlowProducer,
) {}
}
The @InjectFlowProducer() decorator identifies the flow producer by its name, as provided in registerFlowProducer().
Create a flow
async processVideo(videoId: string) {
const flow = await this.flowProducer.add({
name: 'process-video',
queueName: 'video',
data: { videoId },
children: [
{
name: 'extract-audio',
data: { videoId },
queueName: 'audio',
},
{
name: 'generate-thumbnail',
data: { videoId },
queueName: 'image',
},
{
name: 'transcode',
data: { videoId, format: 'mp4' },
queueName: 'video',
},
],
});
return flow;
}
Bulk Job Operations
Add multiple jobs efficiently:
import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
@Injectable()
export class BatchService {
constructor(@InjectQueue('batch') private batchQueue: Queue) {}
async processBatch(items: any[]) {
const jobs = items.map((item, index) => ({
name: 'process',
data: item,
opts: {
jobId: `batch-${Date.now()}-${index}`,
},
}));
const results = await this.batchQueue.addBulk(jobs);
return {
total: results.length,
jobIds: results.map(job => job.id),
};
}
}
Controller Example
Integrate producers with NestJS controllers:
import { Controller, Post, Body, Get, Param } from '@nestjs/common';
import { AudioService } from './audio.service';
@Controller('audio')
export class AudioController {
constructor(private readonly audioService: AudioService) {}
@Post('convert')
async convert(
@Body() body: { file: string; format: string }
) {
const jobId = await this.audioService.convertAudio(
body.file,
body.format
);
return {
message: 'Job queued successfully',
jobId,
};
}
@Get('status/:jobId')
async getStatus(@Param('jobId') jobId: string) {
const status = await this.audioService.getJobStatus(jobId);
if (!status) {
return { error: 'Job not found' };
}
return status;
}
}
Best Practices
Job IDs
Use custom job IDs for deduplication:await queue.add('task', data, {
jobId: `unique-${userId}`,
});
Error Handling
Always handle errors when adding jobs:try {
await queue.add('task', data);
} catch (error) {
// Handle error
}
Job Options
Set appropriate timeouts and attempts:await queue.add('task', data, {
attempts: 3,
backoff: 1000,
});
Auto-removal
Clean up completed jobs automatically:await queue.add('task', data, {
removeOnComplete: true,
});
NestJS Integration
Learn about NestJS integration basics
Queue Events
Listen to queue events in NestJS
External Documentation
NestJS Queues Documentation
Official NestJS documentation for queue integration
