Черги повідомлень

Пояснить роботу черг, буферизацію навантаження, розподіл задач і відокремлення відправників від отримувачів.

Що таке черга повідомлень

Черга повідомлень — це посередник між компонентами системи, який зберігає повідомлення до моменту, коли їх обробить отримувач.

Замість прямої взаємодії:

text

Сервіс замовлень → Сервіс email

сервіси взаємодіють через чергу:

text

Сервіс замовлень → Черга → Сервіс email

text

                  ┌──────────┐
Producer ────────►│  Queue   │
                  └────┬─────┘
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
      Worker 1      Worker 2      Worker 3

text

Queue ── повідомлення ──► Consumer
                           │
                           ├─ помилка → повідомлення залишається
                           │
                           └─ успіх ──► ACK → повідомлення видаляється

text

messageId = payment-123

Якщо payment-123 вже оброблено:
    не виконувати операцію повторно
Інакше:
    виконати операцію
    зберегти payment-123 як оброблений

javascript

class MessageQueue {
  constructor() {
    this.messages = [];
    this.waitingConsumers = [];
  }

  publish(message) {
    const consumer = this.waitingConsumers.shift();

    if (consumer) {
      consumer(message);
      return;
    }

    this.messages.push(message);
  }

  consume() {
    return new Promise((resolve) => {
      const message = this.messages.shift();

      if (message) {
        resolve(message);
        return;
      }

      this.waitingConsumers.push(resolve);
    });
  }

  acknowledge(messageId) {
    console.log(`Повідомлення ${messageId} успішно підтверджено`);
  }
}

const queue = new MessageQueue();

async function processMessage(workerName, message) {
  console.log(`${workerName} обробляє ${message.id}`);

  // Імітуємо тимчасову помилку для першої спроби
  if (message.failOnce && message.attempts === 0) {
    message.attempts += 1;
    throw new Error("Тимчасова помилка зовнішнього сервісу");
  }

  await new Promise((resolve) => setTimeout(resolve, 300));

  queue.acknowledge(message.id);
  console.log(`${workerName} завершив ${message.id}`);
}

async function startWorker(workerName) {
  while (true) {
    const message = await queue.consume();

    try {
      await processMessage(workerName, message);
    } catch (error) {
      console.log(`${workerName}: ${error.message}`);

      if (message.attempts < 3) {
        console.log(`Повторна спроба для ${message.id}`);
        queue.publish(message);
      } else {
        console.log(`Повідомлення ${message.id} перенесено в DLQ`);
      }
    }
  }
}

startWorker("Worker 1");
startWorker("Worker 2");

for (let index = 1; index <= 5; index += 1) {
  queue.publish({
    id: `task-${index}`,
    attempts: 0,
    failOnce: index === 3
  });
}

text

Producer створює повідомлення
        ↓
Queue зберігає повідомлення
        ↓
Consumer отримує повідомлення
        ↓
Consumer обробляє задачу
        ↓
Успіх ─────────────► ACK і видалення
        │
        └─ Помилка ─► повторна спроба
                           │
                           └─ ліміт спроб перевищено → DLQ

Черги повідомлень

Що таке черга повідомлень

Навіщо потрібні черги

Відокремлення компонентів

Буферизація навантаження

Асинхронна обробка

Основні учасники

Producer

Queue

Consumer

Розподіл задач між worker-ами

Підтвердження обробки

Ідемпотентність

Повторні спроби та невдалі повідомлення

Backpressure

Приклад простої черги на Node.js

Порядок повідомлень

Моніторинг черги

Типовий життєвий цикл повідомлення

Типові помилки

Видалення повідомлення до завершення обробки

Відсутність ідемпотентності

Нескінченні повторні спроби

Надто великі повідомлення

Відсутність обмеження черги

Припущення про порядок

Підсумок