Bull для JavaScript (Node.js): полное руководство с примерами

Что такое Bull и зачем он нужен?

Bull — это мощная и популярная библиотека для управления очередями задач (job queues) в среде Node.js. Она построена поверх Redis и предоставляет надежный, масштабируемый и гибкий способ обработки фоновых задач. Если вашему приложению нужно выполнять операции, которые не должны блокировать основной поток (например, отправка email, обработка изображений, генерация PDF, синхронизация с внешними API, парсинг данных), Bull — идеальное решение.

Зачем использовать Bull?

• Асинхронная обработка: Разгружает основной сервер, перемещая тяжелые задачи в фоновый процесс.
• Надежность: Задачи хранятся в Redis, поэтому они не теряются при перезапуске приложения или сбое воркера.
• Повторные попытки: Встроенная поддержка автоматических повторных попыток при ошибках.
• Планирование: Возможность откладывать задачи или запускать их по расписанию (cron).
• Приоритеты: Управление порядком выполнения задач.
• Мониторинг: Легко отслеживать состояние очередей с помощью встроенного UI (Bull Board).

Установка Bull

Для начала работы убедитесь, что у вас установлен и запущен Redis. Bull не работает без него.

1. Установка Redis (пример для Ubuntu/Debian):

   sudo apt update
   sudo apt install redis-server
   sudo systemctl start redis

2. Установка Bull через npm:

   npm install bull

3. Установка Bull Board (опционально, для веб-интерфейса):

   npm install @bull-board/express @bull-board/api

Быстрый старт: первый пример

Создадим простую очередь для отправки приветственных email.

Шаг 1: Создаем продюсера (producer.js)

const Queue = require('bull');

// Создаем очередь с именем 'emailQueue'
const emailQueue = new Queue('emailQueue', {
    redis: { host: 'localhost', port: 6379 }
});

// Функция для добавления задачи в очередь
async function addEmailJob(to, subject, body) {
    const job = await emailQueue.add(
        { to, subject, body }, // Данные задачи
        {
            attempts: 3, // Количество попыток при ошибке
            backoff: { type: 'fixed', delay: 5000 } // Задержка между попытками (5 сек)
        }
    );
    console.log(`Задача добавлена с ID: ${job.id}`);
    return job;
}

// Пример использования
addEmailJob('user@example.com', 'Добро пожаловать!', 'Спасибо за регистрацию.')
    .then(() => process.exit(0))
    .catch(err => console.error(err));

Шаг 2: Создаем воркер (worker.js)

const Queue = require('bull');

// Подключаемся к той же очереди
const emailQueue = new Queue('emailQueue', {
    redis: { host: 'localhost', port: 6379 }
});

// Обработчик задач
emailQueue.process(async (job) => {
    console.log(`Обрабатываю задачу ${job.id}...`);
    const { to, subject, body } = job.data;

    // Имитация отправки email (в реальности используйте nodemailer)
    await new Promise(resolve => setTimeout(resolve, 2000));
    console.log(`Email отправлен на ${to}: ${subject}`);

    // Если что-то пошло не так, выбросьте ошибку
    // throw new Error('Ошибка отправки');
});

console.log('Воркер запущен и ожидает задачи...');

Шаг 3: Запускаем

Сначала запустите воркер: node worker.js. Затем в другом терминале запустите продюсера: node producer.js. Вы увидите, как задача попадает в очередь и обрабатывается.

Основные методы и функции Bull

1. new Queue(name, [options]) — Создание очереди

Что делает: Создает новый экземпляр очереди. Если очередь с таким именем уже существует в Redis, он подключается к ней.

Параметры:

• name (string) — Уникальное имя очереди.
• options (object, опционально) — Настройки:
  • redis (object) — Параметры подключения к Redis (host, port, password, db).
  • defaultJobOptions (object) — Опции по умолчанию для всех задач (attempts, backoff, delay и т.д.).
  • limiter (object) — Ограничение скорости обработки (max, duration).

Пример:

const imageQueue = new Queue('imageProcessing', {
    redis: { host: 'redis.example.com', port: 6380, password: 'secret' },
    defaultJobOptions: { attempts: 5, removeOnComplete: true }
});

2. queue.add(data, [options]) — Добавление задачи

Что делает: Добавляет новую задачу в очередь.

Параметры:

• data (object) — Данные, которые будут переданы воркеру.
• options (object, опционально) — Индивидуальные настройки задачи:
  • delay (number) — Задержка в миллисекундах перед выполнением.
  • attempts (number) — Максимальное количество попыток.
  • backoff (object) — Стратегия повторных попыток (fixed, exponential).
  • priority (number) — Приоритет (чем меньше число, тем выше приоритет).
  • removeOnComplete (boolean) — Удалить задачу после успешного выполнения.
  • removeOnFail (boolean) — Удалить задачу после всех неудачных попыток.

Пример:

const job = await myQueue.add(
    { userId: 123, action: 'generate_report' },
    { delay: 60000, attempts: 3, priority: 1 }
);

3. queue.process(callback) — Обработка задач

Что делает: Регистрирует функцию-обработчик для задач из очереди. Воркер будет вызывать этот callback для каждой задачи.

Параметры:

• callback (function) — Асинхронная функция, которая получает объект job.

Пример:

myQueue.process(async (job) => {
    console.log(`Обработка задачи ${job.id}:`, job.data);
    // Ваша бизнес-логика
    await someAsyncOperation(job.data);
    // Если нужно сообщить о прогрессе
    await job.progress(50);
    await someMoreAsyncOperation(job.data);
    await job.progress(100);
});

4. job.progress(value) — Обновление прогресса

Что делает: Обновляет прогресс выполнения задачи (число от 0 до 100). Полезно для длительных операций.

Параметры:

• value (number) — Текущий прогресс.

Пример:

// Внутри process
for (let i = 0; i < 100; i++) {
    await job.progress(i);
    // ... какая-то работа
}

5. queue.pause() и queue.resume() — Пауза и возобновление

Что делает: Приостанавливает или возобновляет обработку задач в очереди. Задачи продолжают накапливаться, но не обрабатываются.

Параметры: Нет.

Пример:

// Приостановить обработку
await myQueue.pause();
console.log('Очередь приостановлена');

// Через некоторое время возобновить
setTimeout(async () => {
    await myQueue.resume();
    console.log('Очередь возобновлена');
}, 5000);

6. queue.getWaiting() — Получение ожидающих задач

Что делает: Возвращает массив задач, которые ожидают обработки.

Параметры: Нет.

Пример:

const waitingJobs = await myQueue.getWaiting();
console.log(`Ожидающих задач: ${waitingJobs.length}`);
waitingJobs.forEach(job => {
    console.log(`Задача ${job.id}: ${JSON.stringify(job.data)}`);
});

7. queue.getActive() — Получение активных задач

Что делает: Возвращает массив задач, которые в данный момент обрабатываются воркерами.

Параметры: Нет.

Пример:

const activeJobs = await myQueue.getActive();
console.log(`Активных задач: ${activeJobs.length}`);

8. queue.getCompleted() и queue.getFailed() — Завершенные и упавшие задачи

Что делает: Возвращают массивы успешно выполненных или упавших задач (с учетом настроек removeOnComplete и removeOnFail).

Параметры: Нет.

Пример:

const completed = await myQueue.getCompleted();
const failed = await myQueue.getFailed();
console.log(`Успешно: ${completed.length}, Ошибок: ${failed.length}`);

9. queue.removeJobs(pattern) — Удаление задач по шаблону

Что делает: Удаляет все задачи, ID которых соответствуют заданному шаблону (glob pattern).

Параметры:

• pattern (string) — Шаблон для поиска ID задач (например, *report*).

Пример:

// Удалить все задачи, содержащие 'temp' в ID
await myQueue.removeJobs('*temp*');
console.log('Все временные задачи удалены');

10. queue.clean(grace, type) — Очистка старых задач

Что делает: Удаляет задачи старше указанного периода.

Параметры:

• grace (number) — Время в миллисекундах, которое нужно сохранять задачи после их завершения.
• type (string) — Тип задач для очистки: 'completed', 'failed', 'delayed', 'wait'.

Пример:

// Удалить все успешно выполненные задачи старше 1 часа
await myQueue.clean(3600000, 'completed');
console.log('Старые задачи очищены');

11. queue.addRepeatable(name, data, repeatOptions) — Повторяющиеся задачи

Что делает: Создает задачу, которая будет выполняться по расписанию (cron).

Параметры:

• name (string) — Имя задачи (для идентификации).
• data (object) — Данные задачи.
• repeatOptions (object) — Настройки повторения:
  • cron (string) — Выражение cron (например, '0 8 *