0Pricing
Node.js Backend Development Bootcamp · Ders

Üreticiler, Tüketiciler ve AMQP Modeli

AMQP protokolünü kullanarak bir aracıyı bağlayın ve iletileri değiş tokuşlar, kuyruklar ve bağlamalar üzerinden taşıyın.

Üreticiler, Tüketiciler ve AMQP Modeli, CoddyKit'te ücretsiz bir Node.js Backend Development Bootcamp dersidir. Bu, 4 dersinin 1. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, Node.js Backend Development Bootcamp öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. Node.js Backend Development Bootcamp kursu toplamda 4 dersten oluşur.

Bu dersin bazı bölümleri henüz çevrilmemiş olup İngilizce olarak gösterilmektedir.

Why Message Queues?

In a Node.js backend, calling another service directly couples the two together: if the receiver is slow or down, the caller blocks or fails. A message queue decouples them. The sender drops a message into a broker and moves on; the receiver picks it up whenever it is ready.

  • Asynchronous — the producer never waits for the consumer.
  • Resilient — messages survive while a consumer is offline.
  • Scalable — add more consumers to drain a backlog faster.

RabbitMQ is a popular broker that speaks AMQP 0-9-1, the protocol we use throughout this lesson.

The AMQP Model

AMQP separates the act of publishing from the act of storing. A producer never writes to a queue directly. Instead it publishes to an exchange, and the exchange routes the message to one or more queues based on bindings.

  • Producer — publishes messages.
  • Exchange — receives messages and decides where they go.
  • Binding — a rule linking an exchange to a queue (often with a routing key).
  • Queue — buffers messages until a consumer reads them.
  • Consumer — subscribes to a queue and processes messages.

This indirection is what makes routing flexible: change bindings, not producer code.

Connecting from Node.js

The amqplib library is the standard AMQP client for Node.js. You open a TCP connection to the broker, then create a channel over it. A channel is a lightweight virtual connection where almost all AMQP operations happen, so you do not open a new TCP socket per task.

Connection strings use the amqp:// scheme: amqp://user:pass@host:5672/vhost. Always await the connection and handle errors.

const amqp = require('amqplib');

async function connect() {
  const url = process.env.AMQP_URL || 'amqp://guest:guest@localhost:5672';
  const connection = await amqp.connect(url);
  const channel = await connection.createChannel();
  console.log('Connected and channel opened');
  return { connection, channel };
}

connect().catch((err) => {
  console.error('AMQP connection failed:', err.message);
  process.exit(1);
});

Declaring a Queue

Before sending, both sides usually declare the queue. Declaration is idempotent: it creates the queue if missing, otherwise checks the settings match. The most important option is durable.

  • durable: true — the queue definition survives a broker restart.
  • durable: false — the queue is lost on restart.

Durability of the queue is separate from durability of the messages inside it — we will cover persistent messages soon.

async function setupQueue(channel) {
  const queue = 'tasks';
  await channel.assertQueue(queue, { durable: true });
  console.log(`Queue "${queue}" is ready`);
  return queue;
}

The Default Exchange

RabbitMQ ships with a nameless default exchange (the empty string ''). It has a special rule: it automatically binds every queue using the queue's own name as the routing key. So channel.sendToQueue('tasks', ...) is really publishing to the default exchange with routing key 'tasks'.

This is the simplest way to get started, but it hides the exchange concept. Real applications declare their own exchanges for explicit routing, which we do later.

A Producer

A producer publishes a message and then can close the connection. AMQP message bodies are raw bytes, so we serialize to JSON and wrap it in a Buffer. The persistent: true option marks the message so it can be written to disk in a durable queue.

Note the small delay before closing: sendToQueue buffers locally, so we wait for the channel to flush before exiting.

const amqp = require('amqplib');

async function publish() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  const queue = 'tasks';
  await channel.assertQueue(queue, { durable: true });

  const job = { id: 42, type: 'resize-image', file: 'cat.png' };
  channel.sendToQueue(queue, Buffer.from(JSON.stringify(job)), {
    persistent: true,
  });
  console.log('Sent job', job.id);

  await channel.close();
  await conn.close();
}

publish().catch(console.error);

A Consumer

A consumer subscribes to a queue with channel.consume. The broker pushes messages to the callback as they arrive; the consumer typically stays running. The message arrives as msg.content, a Buffer you decode back into your object.

By default consume auto-acknowledges. In the next scene we turn that off so we control exactly when a message is considered done.

const amqp = require('amqplib');

async function consume() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  const queue = 'tasks';
  await channel.assertQueue(queue, { durable: true });

  console.log('Waiting for messages...');
  await channel.consume(queue, (msg) => {
    if (!msg) return;
    const job = JSON.parse(msg.content.toString());
    console.log('Processing job', job.id, job.type);
  });
}

consume().catch(console.error);

Acknowledgements

An acknowledgement (ack) tells the broker a message was successfully handled and can be deleted. Set { noAck: false } and call channel.ack(msg) only after your work finishes.

  • If the consumer crashes before acking, RabbitMQ requeues the message for another consumer.
  • channel.nack(msg, false, true) rejects and requeues; nack(msg, false, false) discards (or dead-letters).

Manual acks are the foundation of at-least-once delivery — never ack before the work is done.

await channel.consume(
  queue,
  async (msg) => {
    if (!msg) return;
    try {
      const job = JSON.parse(msg.content.toString());
      await processJob(job); // your real work
      channel.ack(msg); // success
    } catch (err) {
      channel.nack(msg, false, false); // discard / dead-letter
    }
  },
  { noAck: false }
);

Fair Dispatch with Prefetch

By default RabbitMQ round-robins messages to consumers without considering how busy each one is. A consumer stuck on a slow job can pile up a backlog while others sit idle.

channel.prefetch(1) fixes this: the broker will not send a new message to a consumer until it has acked the previous one. This gives fair dispatch — work flows to whichever consumer is actually free.

async function worker() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  await channel.assertQueue('tasks', { durable: true });

  // Only one unacked message at a time per consumer
  await channel.prefetch(1);

  await channel.consume('tasks', async (msg) => {
    await handle(JSON.parse(msg.content.toString()));
    channel.ack(msg);
  }, { noAck: false });
}

Custom Exchanges and Bindings

For real routing you declare your own exchange and bind queues to it. Exchange types decide the routing logic:

  • direct — exact routing-key match.
  • fanout — broadcast to every bound queue, ignoring the key.
  • topic — wildcard pattern match (e.g. order.*).
  • headers — match on message headers.

Below, an orders direct exchange routes order.created messages to a queue. The producer publishes with channel.publish(exchange, routingKey, content).

async function setupRouting(channel) {
  const exchange = 'orders';
  await channel.assertExchange(exchange, 'direct', { durable: true });

  const queue = 'order_processing';
  await channel.assertQueue(queue, { durable: true });
  await channel.bindQueue(queue, exchange, 'order.created');

  const event = { orderId: 1001, total: 79.9 };
  channel.publish(
    exchange,
    'order.created',
    Buffer.from(JSON.stringify(event)),
    { persistent: true }
  );
}

Putting It Together: A Runnable Demo

Here is a self-contained simulation of the AMQP flow with no broker required — it models an exchange routing to a queue and a consumer draining it. It illustrates the mental model: producer to exchange, exchange to queue via binding, queue to consumer with ack.

Run it to see how the routing key selects the destination queue and how each message is acknowledged exactly once.

// In-memory model of the AMQP routing flow (no broker needed)
class Broker {
  constructor() { this.queues = {}; this.bindings = {}; }
  assertQueue(q) { this.queues[q] = this.queues[q] || []; }
  bind(exchange, queue, key) {
    (this.bindings[exchange] ||= []).push({ queue, key });
  }
  publish(exchange, routingKey, body) {
    for (const b of this.bindings[exchange] || []) {
      if (b.key === routingKey) this.queues[b.queue].push(body);
    }
  }
  consume(queue, handler) {
    let msg;
    while ((msg = this.queues[queue].shift())) handler(msg, () => {});
  }
}

const broker = new Broker();
broker.assertQueue('order_processing');
broker.bind('orders', 'order_processing', 'order.created');

broker.publish('orders', 'order.created', { orderId: 1001 });
broker.publish('orders', 'order.deleted', { orderId: 1002 }); // no binding

broker.consume('order_processing', (msg, ack) => {
  console.log('Consumed:', msg);
  ack();
});

Quick Check

You run several worker consumers on one durable queue. Some jobs take much longer than others, and you notice fast workers sit idle while a few workers are swamped. Which single change best balances the load?

Recap

You connected to a broker and moved messages through the AMQP model.

  • Connection vs channel — one TCP connection, many lightweight channels.
  • Producer to exchange to queue — producers never write queues directly; exchanges route via bindings and routing keys.
  • Exchange types — direct, fanout, topic, headers select the routing logic.
  • Durability and persistence — durable queues plus persistent messages survive restarts.
  • Acks — manual ack/nack after the work gives at-least-once delivery; unacked messages are requeued.
  • Prefetch — prefetch(1) enables fair dispatch across competing consumers.

Next you can explore dead-letter exchanges, message TTLs, and publisher confirms for reliable delivery.

Sıkça Sorulan Sorular

“Üreticiler, Tüketiciler ve AMQP Modeli” dersi ücretsiz mi?

Evet — “Üreticiler, Tüketiciler ve AMQP Modeli” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve Node.js Backend Development Bootcamp kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. Node.js Backend Development Bootcamp kursu toplamda 4 dersten oluşur.

“Üreticiler, Tüketiciler ve AMQP Modeli” dersinde ne öğreneceğim?

AMQP protokolünü kullanarak bir aracıyı bağlayın ve iletileri değiş tokuşlar, kuyruklar ve bağlamalar üzerinden taşıyın. Node.js Backend Development Bootcamp ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

Node.js Backend Development Bootcamp öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te Node.js Backend Development Bootcamp, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 1. dersidir.

“Üreticiler, Tüketiciler ve AMQP Modeli” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu Node.js Backend Development Bootcamp dersinde kod yazıp çalıştırabilir miyim?

Evet. Her Node.js Backend Development Bootcamp dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

  1. Üreticiler, Tüketiciler ve AMQP Modeli
  2. Değiş Tokuş Türleri: Doğrudan, Konu, Yayın ve Üst Bilgiler
  3. Onaylar, Ölü İleti Kuyrukları ve Yeniden Denemeler
  4. İş Kuyrukları, Önceden Getirme ve Rekabet Eden Tüketiciler
← Node.js Backend Development Bootcamp Sayfasına Dön