Что такое Queue?

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

Основной принцип

Очередь работает как реальная очередь в магазине:

• Люди встают в очередь (enqueue — добавить в конец)
• Первый в очереди обслуживается (dequeue — удалить с начала)

// Простая реализация Queue
class Queue {
  constructor() {
    this.items = [];
  }
  
  enqueue(element) {
    this.items.push(element); // добавить в конец
  }
  
  dequeue() {
    return this.items.shift(); // удалить с начала
  }
  
  peek() {
    return this.items[0]; // посмотреть первый элемент
  }
  
  isEmpty() {
    return this.items.length === 0;
  }
  
  size() {
    return this.items.length;
  }
}

const queue = new Queue();
queue.enqueue(1);
queue.enqueue(2);
queue.enqueue(3);

console.log(queue.dequeue()); // 1
console.log(queue.dequeue()); // 2
console.log(queue.dequeue()); // 3

Queue в Node.js

1. Обработка задач с Bull Queue (RabbitMQ):

const Queue = require('bull');

// Создаём очередь
const emailQueue = new Queue('send-emails', {
  redis: { host: '127.0.0.1', port: 6379 }
});

// Добавляем задачу в очередь
await emailQueue.add(
  { email: 'user@example.com', subject: 'Hello' },
  { priority: 10 }
);

// Обрабатываем задачи из очереди
emailQueue.process(async (job) => {
  console.log('Processing:', job.data);
  // Отправляем email
  await sendEmail(job.data);
});

// Слушаем события
emailQueue.on('completed', (job) => {
  console.log('Job completed:', job.id);
});

2. Кьюирование HTTP запросов:

class RequestQueue {
  constructor(maxConcurrent = 3) {
    this.queue = [];
    this.maxConcurrent = maxConcurrent;
    this.processing = 0;
  }
  
  async add(url) {
    return new Promise((resolve) => {
      this.queue.push({ url, resolve });
      this.process();
    });
  }
  
  async process() {
    while (this.processing < this.maxConcurrent && this.queue.length > 0) {
      this.processing++;
      const { url, resolve } = this.queue.shift();
      
      try {
        const response = await fetch(url);
        resolve(response);
      } finally {
        this.processing--;
        this.process();
      }
    }
  }
}

const rq = new RequestQueue(3);
await rq.add('https://api.example.com/data1');
await rq.add('https://api.example.com/data2');

3. Обработка событий WebSocket:

const ws = require('ws');

class MessageQueue {
  constructor() {
    this.queue = [];
    this.processing = false;
  }
  
  addMessage(message) {
    this.queue.push(message);
    this.processQueue();
  }
  
  async processQueue() {
    if (this.processing || this.queue.length === 0) return;
    
    this.processing = true;
    
    while (this.queue.length > 0) {
      const message = this.queue.shift();
      await this.handleMessage(message);
    }
    
    this.processing = false;
  }
  
  async handleMessage(message) {
    console.log('Processing:', message);
    // Обработка
    await new Promise(r => setTimeout(r, 100));
  }
}

Queue vs Stack

Queue (FIFO) — очередь:

• Добавить в конец, удалить с начала
• Для асинхронных задач, обработки событий

Stack (LIFO) — стек:

• Добавить в конец, удалить с конца
• Для отката операций, вложенных вызовов

// Queue
const queue = [];
queue.push(1); // [1]
queue.push(2); // [1, 2]
queue.shift(); // [2], вернул 1

// Stack
const stack = [];
stack.push(1); // [1]
stack.push(2); // [1, 2]
stack.pop(); // [1], вернул 2

Пример: Система уведомлений

class NotificationQueue {
  constructor() {
    this.queue = [];
    this.processing = false;
  }
  
  async sendNotification(userId, message) {
    this.queue.push({ userId, message, timestamp: Date.now() });
    
    if (!this.processing) {
      this.processQueue();
    }
  }
  
  async processQueue() {
    this.processing = true;
    
    while (this.queue.length > 0) {
      const notification = this.queue.shift();
      
      try {
        await this.send(notification);
      } catch (error) {
        console.error('Failed to send:', error);
        // Можем добавить обратно в очередь с retry логикой
      }
    }
    
    this.processing = false;
  }
  
  async send(notification) {
    // Отправляем через WebSocket, Email, SMS и т.д.
    console.log(`Sending to ${notification.userId}: ${notification.message}`);
    await new Promise(r => setTimeout(r, 500));
  }
}

const notificationQueue = new NotificationQueue();
await notificationQueue.sendNotification('user1', 'Hello!');
await notificationQueue.sendNotification('user2', 'Goodbye!');

Когда использовать Queue

• Асинхронная обработка — отправка писем, уведомлений
• Ограничение нагрузки — контроль одновременных операций
• Retry логика — повторная попытка при ошибке
• Распределённые системы — RabbitMQ, Redis Queue
• WebSocket события — управление сообщениями
• Обработка файлов — конвертация, сжатие
• API rate limiting — соблюдение лимитов запросов

Queue — это одна из самых полезных структур в бэкенд-разработке. Она помогает управлять сложными асинхронными потоками и обеспечивает надёжную обработку данных.