Какой принцип работы у queue?

Принцип работы очередей (Queue)

Очереди — FIFO (First In First Out) структуры данных для асинхронной обработки.

FIFO Принцип

[1] [2] [3] [4] [5]
↑                  ↑
Отвод (Dequeue)   Вход (Enqueue)

Простая реализация

class Queue<T> {
  private items: T[] = [];
  
  enqueue(item: T): void { this.items.push(item); }
  dequeue(): T | undefined { return this.items.shift(); }
  peek(): T | undefined { return this.items[0]; }
  size(): number { return this.items.length; }
}

RabbitMQ

import amqp from 'amqplib';

// Отправить
const channel = await connection.createChannel();
await channel.assertQueue('task_queue', { durable: true });
channel.sendToQueue('task_queue', Buffer.from(JSON.stringify(task)), { persistent: true });

// Получить
channel.consume('task_queue', async (msg) => {
  const data = JSON.parse(msg.content.toString());
  await processTask(data);
  channel.ack(msg);
});

Redis Queue (Bull)

import Queue from 'bull';

const emailQueue = new Queue('email');

await emailQueue.add(
  { email: 'user@example.com' },
  { attempts: 3, backoff: { type: 'exponential', delay: 2000 } }
);

emailQueue.process(async (job) => {
  await sendEmail(job.data.email);
  return { success: true };
});

Kafka (Event Streaming)

const producer = kafka.producer();
await producer.send({
  topic: 'user-events',
  messages: [{ key: userId, value: JSON.stringify(event) }],
});

const consumer = kafka.consumer({ groupId: 'email-group' });
await consumer.subscribe({ topic: 'user-events' });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value.toString());
    if (event.type === 'user.created') await sendWelcomeEmail(event);
  },
});

Сравнение

Паттерны

Асинхронная обработка:

await emailQueue.add({ email }); // Сразу возвращаем, обработка в background

Rate limiting:

const queue = new Queue('api', {
  limiter: { max: 100, duration: 60000 }, // 100 за 60 сек
});

Dead Letter Queue для ошибок:

emailQueue.on('failed', async (job, error) => {
  await dlq.add(job.data); // Отправить в очередь ошибок
});

Best Practices

1. Идемпотентность — обработчик безопасно срабатывает дважды
2. Повторы — exponential backoff при ошибках
3. Мониторинг — трекируй размер очереди
4. Таймауты — устанавливай для обработчиков
5. Логирование — логируй каждую задачу

Выбор очереди

• Redis (Bull) — для большинства Node.js приложений
• RabbitMQ — микросервисы, гарантии доставки
• Kafka — event streaming, масштабирование
• In-Memory — development, простые очереди