Message Queues (RabbitMQ & Kafka)

const amqp = require('amqplib');

const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
await channel.assertQueue('orders');

// Publisher
channel.sendToQueue('orders', Buffer.from(JSON.stringify({ orderId: 123 })));

// Consumer
channel.consume('orders', (msg) => {
  const order = JSON.parse(msg.content.toString());
  processOrder(order);
  channel.ack(msg); // acknowledge only after successful processing
});

Common follow-ups: What happens to an unacknowledged message if the consumer that received it crashes before calling ack()?;How do RabbitMQ exchanges (direct, topic, fanout) differ in how they route published messages to queues?

const { Kafka } = require('kafkajs');

const kafka = new Kafka({ clientId: 'my-app', brokers: ['localhost:9092'] });
const producer = kafka.producer();
await producer.send({ topic: 'orders', messages: [{ key: String(orderId), value: JSON.stringify(order) }] });

const consumer = kafka.consumer({ groupId: 'order-processors' });
await consumer.subscribe({ topic: 'orders' });
await consumer.run({ eachMessage: async ({ message }) => processOrder(JSON.parse(message.value)) });

Common follow-ups: Why is message ordering only guaranteed within a single Kafka partition, not across an entire topic?;How does the choice of a message's partition key affect both ordering guarantees and load distribution across consumers?

// Risky: auto-ack, message is 'confirmed received' before processing even starts
channel.consume('orders', processOrder, { noAck: true });

// Reliable: manual ack, only confirmed after processing genuinely succeeds
channel.consume('orders', async (msg) => {
  await processOrder(JSON.parse(msg.content.toString()));
  channel.ack(msg); // only now does the broker consider it handled
}, { noAck: false });

Common follow-ups: What's the tradeoff of manual acknowledgment in terms of the possibility of processing the same message twice (at-least-once rather than exactly-once)?;How does this relate back to the job idempotency concept discussed for background job queues?

await channel.assertExchange('dlx', 'direct');
await channel.assertQueue('failed-orders');
await channel.bindQueue('failed-orders', 'dlx', '');

await channel.assertQueue('orders', {
  arguments: { 'x-dead-letter-exchange': 'dlx' },
});

channel.consume('orders', (msg) => {
  try {
    processOrder(msg);
    channel.ack(msg);
  } catch (err) {
    channel.nack(msg, false, false); // false, false = don't requeue -- routes to the DLX instead
  }
});

Common follow-ups: How do you distinguish between a transient failure worth retrying automatically versus a permanent failure that should go straight to the dead-letter queue?;What monitoring would you set up to alert when messages start accumulating in a dead-letter queue?

// Tightly coupled: caller blocks and fails if inventory-service is down
await fetch('http://inventory-service/reserve', { method: 'POST', body: JSON.stringify(order) });

// Decoupled: publish and move on; inventory-service processes whenever it's available
await channel.sendToQueue('inventory-reservations', Buffer.from(JSON.stringify(order)));

Common follow-ups: What's the cost of the eventual consistency this decoupling introduces, and in what scenarios is that unacceptable (like needing an immediate inventory check before confirming an order)?;How would you combine synchronous calls for time-sensitive checks with asynchronous messaging for everything else in the same order flow?