Bull Queue — task queue, background jobs

~25 daq · NestJS · Ilg'or mavzular

Muammodan boshlaylik

Foydalanuvchi ro'yxatdan o'tdi. Unga xush kelibsiz email yuborish kerak:

async register(dto: RegisterDto) {
  const user = await this.usersService.create(dto);
  await this.emailService.sendWelcomeEmail(user.email); // ← 2-3 soniya kutadi
  return user;
}

Foydalanuvchi 3 soniya kutdi. Email servisi sekin ishlasa — 10 soniya. Email servisi vaqtincha ishlamasa — ro'yxatdan o'tish ham ishlamaydi.

Ikkita muammo: sekinlik va bog'liqlik.

Queue nima?

Queue (navbat) — vazifalarni navbatga qo'yib, keyinroq bajarish.

Hayotiy o'xshatma

Restoran. Ofitsiant buyurtmani oshpazga topshiradi va darhol keyingi mijozga ketadi. Oshpaz o'z tezligida tayyorlaydi. Ofitsiant oshpaz tugashini kutmaydi.

Foydalanuvchi → API (darhol javob) → Queue → Worker (fonda email yuboradi)

Bull Queue — Redis orqali ishlaydigan NestJS queue kutubxonasi.

O'rnatish

npm install @nestjs/bull bull
npm install -D @types/bull

Redis kerak:

docker run -d -p 6379:6379 redis:alpine

Fayl tuzilmasi

src/
  email/
    email.module.ts     ← BullModule shu yerda sozlanadi
    email.service.ts    ← job navbatga qo'shadi (Producer)
    email.processor.ts  ← jobni bajaradi (Processor)
  app.module.ts         ← BullModule.forRoot() shu yerda

1-qadam: AppModule ga Redis ulash

// app.module.ts
import { BullModule } from '@nestjs/bull';

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

2-qadam: Queue yaratish

// email/email.module.ts
import { BullModule } from '@nestjs/bull';
import { EmailService } from './email.service';
import { EmailProcessor } from './email.processor';

@Module({
  imports: [
    BullModule.registerQueue({ name: 'email' }), // ← 'email' navbati yaratildi
  ],
  providers: [EmailService, EmailProcessor],
})
export class EmailModule {}

3-qadam: Job qo'shish (Producer)

// email/email.service.ts
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Injectable()
export class EmailService {
  constructor(@InjectQueue('email') private emailQueue: Queue) {}

  async sendWelcomeEmail(to: string, name: string) {
    await this.emailQueue.add('welcome', { to, name }); // ← navbatga qo'shildi
    // Shu yerda tugaydi — email fonda yuboriladi
  }
}

Endi register() darhol javob beradi:

async register(dto: RegisterDto) {
  const user = await this.usersService.create(dto);
  await this.emailService.sendWelcomeEmail(user.email, user.name); // ← kutmaydi
  return user; // ← darhol qaytadi
}

4-qadam: Job bajarish (Processor)

// email/email.processor.ts
import { Process, Processor } from '@nestjs/bull';
import { Job } from 'bull';

@Processor('email') // ← 'email' navbatidagi joblarni oladi
export class EmailProcessor {

  @Process('welcome') // ← 'welcome' nomli jobni bajaradi
  async handleWelcomeEmail(job: Job<{ to: string; name: string }>) {
    const { to, name } = job.data;
    console.log(`Email yuborildi: ${to}, ${name} uchun`);
    // nodemailer yoki boshqa kutubxona bilan haqiqiy email yuborish
  }
}

Xato bo'lsa qayta urish

await this.emailQueue.add(
  'welcome',
  { to, name },
  {
    attempts: 3,   // ← xato bo'lsa 3 marta urinib ko'r
    backoff: 5000, // ← urinishlar orasida 5 soniya kut
  },
);

Email serveri vaqtincha ishlamasa ham — Bull 3 marta qayta urinadi. Foydalanuvchining email'i yo'qolmaydi.

Kechiktirib yuborish

// 30 daqiqadan keyin yuborish
await this.emailQueue.add(
  'reminder',
  { to },
  { delay: 30 * 60 * 1000 }, // ← millisoniyada
);

Tez-tez uchraydigan xatolar

1. Redis ulanmayapti

Error: connect ECONNREFUSED 127.0.0.1:6379

Yechim: docker ps bilan Redis ishlayotganini tekshiring.

2. Processor ishlamayapti

No processor defined for job type "welcome"

Sababi: @Process('welcome') nomi va queue.add('welcome') dagi nom mos emas. Yechim: Nomlarni tekshiring — katta-kichik harf muhim.

3. EmailModule AppModule ga qo'shilmagan Yechim: app.module.ts imports ga EmailModule qo'shing.

Xulosa

• Queue — sekin vazifalarni fonga yuborish, foydalanuvchi kutmaydi
• Producer — queue.add() bilan job navbatga qo'shadi
• Processor — @Process() bilan jobni bajaradi
• Redis — joblarni saqlaydi
• attempts + backoff — xato bo'lsa avtomatik qayta urish

Keyingi dars: 42-dars: GraphQL — asoslar va tushunchalar