Налаштування черг повідомлень (BullMQ) для Node.js веб-застосунку

Налаштування BullMQ для чергиці завдань у Node.js

BullMQ — бібліотека чергиці для Node.js на базі Redis. Підтримує пріоритети, затримки, повторні спроби з exponential backoff, cron-розклад, Worker threads. Підходить для більшості завдань: електронна пошта, сповіщення, обробка медіа, генерація PDF.

Встановлення

npm install bullmq ioredis

Конфігурація з'єднання

// lib/redis.ts
import { Redis } from 'ioredis';

export const redisConnection = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: Number(process.env.REDIS_PORT) || 6379,
  password: process.env.REDIS_PASSWORD,
  maxRetriesPerRequest: null,  // обов'язково для BullMQ
  enableReadyCheck: false,
});

Визначення чергиці

// queues/index.ts
import { Queue } from 'bullmq';
import { redisConnection } from '../lib/redis';

const defaultJobOptions = {
  attempts: 3,
  backoff: {
    type: 'exponential' as const,
    delay: 5000,  // 5s, 10s, 20s
  },
  removeOnComplete: { count: 1000, age: 86400 },
  removeOnFail: { count: 5000, age: 604800 },
};

export const emailQueue = new Queue('emails', {
  connection: redisConnection,
  defaultJobOptions,
});

export const notificationQueue = new Queue('notifications', {
  connection: redisConnection,
  defaultJobOptions,
});

export const reportQueue = new Queue('reports', {
  connection: redisConnection,
  defaultJobOptions: {
    ...defaultJobOptions,
    attempts: 1,  // звіти не повторяємо
  },
});

Воркери

// workers/emailWorker.ts
import { Worker, Job } from 'bullmq';
import { redisConnection } from '../lib/redis';
import { sendEmail } from '../services/email';

interface EmailJobData {
  to: string;
  subject: string;
  template: string;
  variables: Record<string, unknown>;
}

const worker = new Worker<EmailJobData>(
  'emails',
  async (job: Job<EmailJobData>) => {
    const { to, subject, template, variables } = job.data;

    await job.updateProgress(10);
    await sendEmail({ to, subject, template, variables });
    await job.updateProgress(100);

    return { sent: true, to, timestamp: new Date().toISOString() };
  },
  {
    connection: redisConnection,
    concurrency: 10,          // 10 паралельних завдань
    limiter: {
      max: 100,               // не більше 100 завдань
      duration: 60_000,       // на хвилину (rate limiting)
    },
  }
);

worker.on('completed', (job, result) => {
  console.log(`Email sent to ${result.to}`);
});

worker.on('failed', (job, err) => {
  console.error(`Email failed: job ${job?.id}:`, err.message);
});

worker.on('error', (err) => {
  console.error('Worker error:', err);
});

export default worker;

Додавання завдань

// Просте завдання
await emailQueue.add('welcome-email', {
  to: user.email,
  subject: 'Ласкаво просимо!',
  template: 'welcome',
  variables: { name: user.name },
});

// З затримкою (5 хвилин)
await emailQueue.add('follow-up-email', {
  to: user.email,
  subject: 'Як дела?',
  template: 'follow-up',
  variables: { name: user.name },
}, {
  delay: 5 * 60 * 1000,
});

// З пріоритетом (1 = найвищий)
await notificationQueue.add('push-notification', {
  userId: user.id,
  message: 'Терміне сповіщення',
}, {
  priority: 1,
});

// Масова відправка
const jobs = users.map(user => ({
  name: 'newsletter',
  data: { to: user.email, template: 'newsletter' },
  opts: { delay: Math.random() * 60_000 },  // розмазати по хвилині
}));
await emailQueue.addBulk(jobs);

Повторювані завдання (Cron)

// Щоденний звіт о 9:00 UTC
await reportQueue.add(
  'daily-report',
  { type: 'daily', recipients: ['[email protected]'] },
  {
    repeat: { pattern: '0 9 * * *' },
    jobId: 'daily-report-unique',  // запобігти дублікатам
  }
);

// Кожні 5 хвилин: health check
await emailQueue.add(
  'health-check',
  {},
  { repeat: { every: 5 * 60 * 1000 } }
);

// Отримати всі повторювані завдання
const repeatableJobs = await reportQueue.getRepeatableJobs();

// Видалити
await reportQueue.removeRepeatableByKey(repeatableJobs[0].key);

BullMQ Board (моніторинг)

// app.ts
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';
import { emailQueue, notificationQueue, reportQueue } from './queues';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(notificationQueue),
    new BullMQAdapter(reportQueue),
  ],
  serverAdapter,
});

app.use('/admin/queues', authenticate, serverAdapter.getRouter());

Flow (залежні завдання)

import { FlowProducer } from 'bullmq';

const flow = new FlowProducer({ connection: redisConnection });

// Створити завдання із залежностями: спочатку resize, потім upload, потім notify
await flow.add({
  name: 'notify-user',
  queueName: 'notifications',
  data: { userId },
  children: [{
    name: 'upload-to-s3',
    queueName: 'uploads',
    data: { tempPath },
    children: [{
      name: 'resize-image',
      queueName: 'images',
      data: { originalPath, sizes: [200, 400, 800] },
    }],
  }],
});

Терміни реалізації

BullMQ для типового Node.js проекту (електронна пошта, сповіщення, cron): 2–3 дні. З BullBoard, моніторингом та Flow: 3–4 дні.