Очереди задач в JavaScript (Node.js): полное руководство

Работа с очередями задач в JavaScript (Node.js)

В современных веб-приложениях и микросервисных архитектурах часто возникает необходимость выполнять фоновые задачи, которые не должны блокировать основной поток выполнения. Очереди задач — это мощный инструмент, который помогает управлять асинхронными операциями, обеспечивая надежность, масштабируемость и отказоустойчивость. В этой статье мы подробно разберем, как работать с очередями задач в среде Node.js, рассмотрим основные библиотеки и напишем практические примеры кода.

Что такое очередь задач и зачем она нужна?

Очередь задач (Task Queue) — это структура данных, работающая по принципу FIFO (First In, First Out — «первым пришёл, первым ушёл»). Задачи помещаются в очередь и обрабатываются воркерами (worker) в порядке поступления. Основные сценарии использования включают:

• Отправка email-уведомлений — не нужно заставлять пользователя ждать, пока письмо будет отправлено.
• Генерация отчетов или PDF — ресурсоемкие операции, которые лучше выполнять асинхронно.
• Обработка загруженных изображений — изменение размера, сжатие, добавление водяных знаков.
• Интеграция с внешними API — выполнение запросов с ограничением по частоте (rate limiting).
• Плановые задачи (cron) — регулярная очистка базы данных или обновление кэша.

Использование очередей позволяет разгрузить основной сервер, повысить отказоустойчивость (если один воркер упадёт, задача будет передана другому) и гарантировать выполнение задачи даже при временных сбоях.

Простая очередь в памяти на Node.js

Для начала реализуем простейшую очередь без внешних зависимостей. Такой подход подходит для прототипирования или очень маленьких проектов, но не рекомендуется для продакшена из-за отсутствия персистентности.

class SimpleQueue {
    constructor() {
        this.queue = [];
        this.processing = false;
    }

    enqueue(task) {
        this.queue.push(task);
        this.processNext();
    }

    async processNext() {
        if (this.processing || this.queue.length === 0) return;
        this.processing = true;
        const task = this.queue.shift();
        try {
            await task();
        } catch (err) {
            console.error('Ошибка выполнения задачи:', err);
        } finally {
            this.processing = false;
            this.processNext();
        }
    }
}

// Пример использования
const queue = new SimpleQueue();

queue.enqueue(async () => {
    console.log('Задача 1: Отправка email');
});

queue.enqueue(async () => {
    console.log('Задача 2: Генерация отчета');
});

console.log('Задачи добавлены в очередь');

Этот код демонстрирует базовую механику: задачи выполняются последовательно, одна за другой. Однако при перезапуске сервера все необработанные задачи будут потеряны.

Bull — профессиональная очередь на Redis

Для серьёзных проектов в Node.js стандартом де-факто является библиотека Bull. Она использует Redis в качестве бэкенда, что обеспечивает персистентность, поддержку приоритетов, повторные попытки и отложенные задачи.

Установка:

npm install bull

Пример простого продюсера и консюмера:

const Queue = require('bull');

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

// Функция-воркер: обрабатывает задачи
emailQueue.process(async (job) => {
    const { to, subject, body } = job.data;
    console.log(`Отправка email на ${to}: ${subject}`);
    // Здесь был бы реальный код отправки email
    await new Promise(resolve => setTimeout(resolve, 1000));
    return { success: true };
});

// Добавление задачи в очередь (продюсер)
async function addEmailTask(to, subject, body) {
    const job = await emailQueue.add(
        { to, subject, body },
        {
            attempts: 3,          // Повторить 3 раза при ошибке
            backoff: 5000,        // Задержка между повторами (5 сек)
            removeOnComplete: true // Удалить задачу после успеха
        }
    );
    console.log(`Задача добавлена, ID: ${job.id}`);
}

// Добавляем несколько задач
addEmailTask('user@example.com', 'Приветствие', 'Добро пожаловать!');
addEmailTask('admin@example.com', 'Отчет', 'Ежедневный отчет готов.');

Ключевые возможности Bull:

• Автоматический повтор при ошибках (retry).
• Отложенные задачи (delay).
• Приоритеты задач.
• Пауза и возобновление очереди.
• События (completed, failed, progress).
• Веб-интерфейс для мониторинга (через отдельный пакет bull-board).

RabbitMQ и amqplib — корпоративный уровень

Для распределенных систем с высокой нагрузкой часто используют RabbitMQ — брокер сообщений, поддерживающий протокол AMQP. В Node.js для работы с ним применяется библиотека amqplib.

Установка:

npm install amqplib

Пример продюсера (отправителя):

const amqp = require('amqplib');

async function sendTask() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'task_queue';

    // Убедимся, что очередь существует (durable = true для сохранения)
    await channel.assertQueue(queueName, { durable: true });

    const message = JSON.stringify({ task: 'generate_report', userId: 123 });
    channel.sendToQueue(queueName, Buffer.from(message), {
        persistent: true // Сохранить сообщение на диск
    });

    console.log('Сообщение отправлено');
    setTimeout(() => {
        connection.close();
        process.exit(0);
    }, 500);
}

sendTask();

Пример консюмера (воркера):

const amqp = require('amqplib');

async function startWorker() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'task_queue';

    await channel.assertQueue(queueName, { durable: true });

    // Гарантируем, что воркер получит только одну задачу за раз
    channel.prefetch(1);

    console.log('Ожидание задач...');
    channel.consume(queueName, async (msg) => {
        if (msg !== null) {
            const data = JSON.parse(msg.content.toString());
            console.log('Получена задача:', data.task);

            // Имитация обработки
            await new Promise(resolve