Настройка очередей сообщений (BullMQ) для Node.js веб-приложения

Настройка BullMQ для очереди задач в Node.js

BullMQ — библиотека очередей для Node.js на базе Redis. Поддерживает приоритеты, задержки, повторные попытки с exponential backoff, cron-расписание, Worker threads. Подходит для большинства задач: emails, уведомления, обработка медиа, 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 проекта (emails, уведомления, cron): 2–3 дня. С BullBoard, мониторингом и Flow: 3–4 дня.