Продюсери та консьюмери · fullstack.org.ua

fullstack.org.ua

Продюсери та консьюмери · fullstack.org.ua

json

{
  "type": "order.created",
  "messageId": "msg-8f31",
  "createdAt": "2026-09-02T10:15:00.000Z",
  "payload": {
    "orderId": "order-123",
    "customerId": "customer-42",
    "total": 1499.99
  }
}

text

messageId = msg-8f31

якщо msg-8f31 уже оброблено:
    надіслати ack
інакше:
    виконати операцію
    зберегти msg-8f31 як оброблене
    надіслати ack

javascript

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

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

    if (consumer) {
      consumer(message);
    } else {
      this.messages.push(message);
    }
  }

  consume(onMessage) {
    const message = this.messages.shift();

    if (message) {
      onMessage(message);
    } else {
      this.waitingConsumers.push(onMessage);
    }
  }

  size() {
    return this.messages.length;
  }
}

const queue = new MessageQueue();
let activeMessages = 0;

const maxConcurrentMessages = 2;

function publishOrder(orderId) {
  const message = {
    messageId: `message-${orderId}`,
    type: "order.created",
    createdAt: new Date().toISOString(),
    payload: {
      orderId,
      total: 1499.99
    }
  };

  queue.publish(message);
  console.log(`Продюсер: надіслано ${message.messageId}`);
}

function startConsumer() {
  if (activeMessages >= maxConcurrentMessages) {
    return;
  }

  activeMessages += 1;

  queue.consume(async (message) => {
    try {
      console.log(`Консьюмер: отримано ${message.messageId}`);

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

      console.log(`Консьюмер: замовлення ${message.payload.orderId} оброблено`);
      console.log(`Консьюмер: ack для ${message.messageId}`);
    } finally {
      activeMessages -= 1;
      startConsumer();
    }
  });
}

for (let i = 1; i <= 5; i += 1) {
  publishOrder(`order-${i}`);
}

const timer = setInterval(() => {
  startConsumer();

  if (queue.size() === 0 && activeMessages === 0) {
    clearInterval(timer);
    console.log("Усі повідомлення оброблено");
  }
}, 100);