Очереди задач в TypeScript: создание и оптимизация обработки

Работа с очередями задач в TypeScript: полное руководство

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

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

Очередь задач — это структура данных, работающая по принципу FIFO (First In, First Out). Задачи добавляются в конец очереди, а извлекаются из начала. В контексте TypeScript и Node.js очереди используются для:

• Ограничения параллелизма — контроль количества одновременно выполняемых операций (например, запросы к API с лимитами).
• Асинхронной обработки — выполнение тяжелых задач в фоне без блокировки основного потока.
• Повторных попыток — автоматический перезапуск упавших задач с задержкой.
• Управления порядком — гарантия последовательного выполнения зависимых операций.

TypeScript добавляет строгую типизацию, что делает код очередей более надежным и предсказуемым.

Базовая реализация очереди задач на TypeScript

Начнем с создания простой, но полностью типизированной очереди. Она будет принимать задачи (функции, возвращающие Promise) и выполнять их последовательно.

// Определяем тип задачи
type Task<T> = () => Promise<T>;

// Класс очереди
class TaskQueue<T> {
    private queue: Task<T>[] = [];
    private isProcessing: boolean = false;

    // Добавление задачи в очередь
    enqueue(task: Task<T>): void {
        this.queue.push(task);
        this.process();
    }

    // Запуск обработки
    private async process(): Promise<void> {
        if (this.isProcessing) return;
        this.isProcessing = true;

        while (this.queue.length > 0) {
            const task = this.queue.shift()!;
            try {
                const result = await task();
                console.log('Задача выполнена:', result);
            } catch (error) {
                console.error('Ошибка выполнения задачи:', error);
            }
        }

        this.isProcessing = false;
    }
}

// Пример использования
const queue = new TaskQueue<string>();
queue.enqueue(async () => {
    await new Promise(resolve => setTimeout(resolve, 1000));
    return 'Первая задача';
});
queue.enqueue(async () => 'Вторая задача (быстрая)');

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

Очередь с ограничением параллелизма

Часто требуется выполнять несколько задач одновременно, но не превышать определенный лимит (например, 5 параллельных запросов). Добавим параметр concurrency.

class ParallelTaskQueue<T> {
    private queue: Task<T>[] = [];
    private activeCount: number = 0;
    private readonly concurrency: number;

    constructor(concurrency: number = 3) {
        this.concurrency = concurrency;
    }

    enqueue(task: Task<T>): void {
        this.queue.push(task);
        this.processNext();
    }

    private processNext(): void {
        while (this.activeCount < this.concurrency && this.queue.length > 0) {
            const task = this.queue.shift()!;
            this.activeCount++;
            task()
                .then(result => console.log('Успех:', result))
                .catch(error => console.error('Ошибка:', error))
                .finally(() => {
                    this.activeCount--;
                    this.processNext();
                });
        }
    }
}

// Использование: максимум 2 параллельные задачи
const parallelQueue = new ParallelTaskQueue<number>(2);
for (let i = 0; i < 5; i++) {
    parallelQueue.enqueue(async () => {
        await new Promise(resolve => setTimeout(resolve, Math.random() * 2000));
        return i;
    });
}

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

Приоритетная очередь задач

В реальных приложениях не все задачи равны. Критические операции (например, обработка платежа) должны выполняться раньше фоновых (например, отправка email-уведомления). Реализуем приоритетную очередь с использованием бинарной кучи или простого массива.

// Тип задачи с приоритетом
interface PriorityTask<T> {
    priority: number; // Чем меньше число, тем выше приоритет
    task: Task<T>;
}

class PriorityQueue<T> {
    private queue: PriorityTask<T>[] = [];
    private isProcessing: boolean = false;

    enqueue(task: Task<T>, priority: number = 0): void {
        this.queue.push({ task, priority });
        // Сортируем по приоритету (высокий приоритет = маленькое число)
        this.queue.sort((a, b) => a.priority - b.priority);
        this.process();
    }

    private async process(): Promise<void> {
        if (this.isProcessing) return;
        this.isProcessing = true;

        while (this.queue.length > 0) {
            const { task, priority } = this.queue.shift()!;
            console.log(`Выполняется задача с приоритетом ${priority}`);
            try {
                await task();
            } catch (error) {
                console.error(`Задача с приоритетом ${priority} упала:`, error);
            }
        }

        this.isProcessing = false;
    }
}

// Пример с разными приоритетами
const priorityQueue = new PriorityQueue<void>();
priorityQueue.enqueue(async () => {
    console.log('Фоновая задача');
}, 10);
priorityQueue.enqueue(async () => {
    console.log('Критическая задача');
}, 1);
priorityQueue.enqueue(async () => {
    console.log('Обычная задача');
}, 5);

Обратите внимание: сортировка при каждом добавлении — не самое эффективное решение для больших очередей. Для продакшена лучше использовать структуру "куча" (heap).

Продвинутая очередь с повторными попытками и таймаутами

Надежная очередь должна уметь перезапускать упавшие задачи и отменять зависшие. Добавим эти возможности.

interface RetryTask<T> {
    task: Task<T>;
    retries: number;
    maxRetries: number;
    timeout: number;
}

class RobustQueue<T> {
    private queue: RetryTask<T>[] = [];
    private isProcessing: boolean = false;

    enqueue