Очереди задач в TypeScript: практическое руководство

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

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

Мы рассмотрим создание простой очереди, работу с приоритетами, обработку ошибок, а также интеграцию с Redis через библиотеку BullMQ. Все примеры будут написаны на TypeScript с акцентом на строгую типизацию.

1. Основы: создаем простую очередь задач

Для начала реализуем простую очередь задач без внешних зависимостей. Это поможет понять базовые принципы работы. Наша очередь будет хранить функции-задачи и выполнять их последовательно.

type Task<T> = () => Promise<T>;

interface TaskItem<T> {
  task: Task<T>;
  resolve: (value: T | PromiseLike<T>) => void;
  reject: (reason?: any) => void;
}

class SimpleQueue {
  private queue: TaskItem<any>[] = [];
  private isProcessing = false;

  add<T>(task: Task<T>): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      this.queue.push({ task, resolve, reject });
      this.process();
    });
  }

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

    const item = this.queue.shift()!;
    try {
      const result = await item.task();
      item.resolve(result);
    } catch (error) {
      item.reject(error);
    } finally {
      this.isProcessing = false;
      this.process();
    }
  }
}

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

queue.add(async () => {
  console.log('Задача 1 началась');
  await new Promise(r => setTimeout(r, 1000));
  console.log('Задача 1 завершена');
  return 'Результат 1';
});

queue.add(async () => {
  console.log('Задача 2 началась');
  return 'Результат 2';
});

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

2. Приоритизация и обработка ошибок

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

type Priority = 'high' | 'medium' | 'low';

interface PriorityTask<T> {
  task: Task<T>;
  priority: Priority;
  resolve: (value: T) => void;
  reject: (reason?: any) => void;
  retries: number;
}

class PriorityQueue {
  private high: PriorityTask<any>[] = [];
  private medium: PriorityTask<any>[] = [];
  private low: PriorityTask<any>[] = [];
  private isProcessing = false;
  private maxRetries = 3;

  add<T>(task: Task<T>, priority: Priority = 'medium'): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      const taskItem: PriorityTask<T> = {
        task,
        priority,
        resolve,
        reject,
        retries: 0
      };
      this.getQueue(priority).push(taskItem);
      this.process();
    });
  }

  private getQueue(priority: Priority): PriorityTask<any>[] {
    switch (priority) {
      case 'high': return this.high;
      case 'medium': return this.medium;
      case 'low': return this.low;
    }
  }

  private getNextTask(): PriorityTask<any> | undefined {
    if (this.high.length > 0) return this.high.shift();
    if (this.medium.length > 0) return this.medium.shift();
    return this.low.shift();
  }

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

    const item = this.getNextTask();
    if (!item) {
      this.isProcessing = false;
      return;
    }

    try {
      const result = await item.task();
      item.resolve(result);
    } catch (error) {
      if (item.retries < this.maxRetries) {
        item.retries++;
        console.log(`Повторная попытка ${item.retries} для задачи с приоритетом ${item.priority}`);
        this.getQueue(item.priority).push(item);
      } else {
        item.reject(error);
      }
    } finally {
      this.isProcessing = false;
      this.process();
    }
  }
}

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

pQueue.add(async () => {
  // Критическая задача
  await someCriticalOperation();
}, 'high');

pQueue.add(async () => {
  // Фоновая задача
  await cleanupOldFiles();
}, 'low');

Мы добавили три очереди по приоритетам и механизм повторных попыток. Если задача завершается ошибкой, она возвращается в конец своей очереди до достижения лимита попыток. Это повышает отказоустойчивость системы.

3. Промышленное решение: BullMQ + TypeScript

Для production-проектов лучше использовать специализированные библиотеки. BullMQ — это мощная очередь задач на базе Redis, которая предоставляет все необходимые функции: сохранение состояния, планирование, повторные попытки, мониторинг и масштабирование. Рассмотрим интеграцию с TypeScript.

import { Queue, Worker, Job } from 'bullmq';
import IORedis from 'ioredis';

// Определяем типы данных для задач
interface EmailTaskData {
  to: string;
  subject: string;
  body: string;
}

interface ReportTaskData {
  userId: string;
  reportType: 'weekly' | 'monthly';
}

// Создаем подключение к Redis
const connection = new IORedis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: null
});

// Инициализируем очереди
const emailQueue = new Queue<EmailTaskData>('email', { connection });
const reportQueue = new Queue<ReportTaskData>('report', { connection });

// Добавляем задачи с опциями
async function addEmailTask(data: EmailTaskData): Promise<Job<EmailTaskData>> {
  return emailQueue.add('send-email', data, {
    attempts: 3,          // Количество попыток
    backoff: {            // Стратегия ожидания между попытками
      type: 'exponential',