Messaging Patterns with AMQP in TypeScript

← Back to the section

Once services start talking through a broker, a question arises: how do you correctly organize queues, exchanges, and subscriptions? Most tasks fit into a handful of standard schemes. Let's walk through each one with amqplib examples.

Hand out tasks to several workers — Work Queue

Picture this: users upload photos, and each one has to be compressed and cropped into several sizes. That takes time. Doing it right inside the HTTP request is a no-go — the user would be waiting minutes.

The solution is a work queue: save the job into a queue and hand it to one of a pool of workers.

The broker distributes the tasks among the workers itself. Each message goes to exactly one of them.

const channel = await connection.createChannel();
await channel.assertQueue('images.to-process', {
  durable: true,
  arguments: { 'x-queue-type': 'quorum' },
});

await channel.prefetch(20);
await channel.consume('images.to-process', async (msg) => {
  if (!msg) return;
  const job: ImageJob = JSON.parse(msg.content.toString());
  // process the image
  channel.ack(msg);
});

prefetch(20) means: the worker holds up to 20 unacknowledged messages at a time. If you run 5 copies of the service, you get up to 100 parallel workers.

When to pick it: background tasks — file processing, sending emails, generating reports, anything that is "drop it in a queue and someone will pick it up".

Send an event to every service at once — Publish/Subscribe

A different task: the event "configuration updated" happened, and every service must refresh its cache. You can't know in advance who exactly is subscribed and how many services are running.

This is publish/subscribe: the publisher sends a single message, and all subscribers receive a copy at the same time.

Here you need a fanout exchange — it copies every message into all bound queues. Each service declares its own queue and binds it to the shared exchange.

await channel.assertExchange('cache.invalidation', 'fanout', { durable: true });

const { queue } = await channel.assertQueue('', {
  exclusive: true,
  autoDelete: true,
  durable: false,
});
await channel.bindQueue(queue, 'cache.invalidation', '');

await channel.consume(queue, (msg) => {
  if (!msg) return;
  const event: CacheInvalidationEvent = JSON.parse(msg.content.toString());
  cache.evict(event.key);
  channel.ack(msg);
});

exclusive + autoDelete — the queue belongs to a single connection and is deleted when it disconnects. On a service restart, no garbage piles up in the broker.

When to pick it: cache invalidation, broadcast notifications to the whole cluster, configuration updates.

Route an event to the right worker — Routing

Sometimes you don't want "everyone", you want "exactly the one who needs it". For example: the order.created event should go to the fulfillment service and to audit, while order.payment-failed should go only to alerts.

This is routing: a direct exchange looks at the message's routing key and delivers it only to queues with a matching binding key.

await channel.assertExchange('orders', 'direct', { durable: true });

const quorum = { durable: true, arguments: { 'x-queue-type': 'quorum' } };
await channel.assertQueue('orders.fulfillment', quorum);
await channel.assertQueue('orders.audit', quorum);
await channel.assertQueue('orders.alerts', quorum);

await channel.bindQueue('orders.fulfillment', 'orders', 'order.created');
await channel.bindQueue('orders.audit', 'orders', 'order.created');
await channel.bindQueue('orders.audit', 'orders', 'order.cancelled');
await channel.bindQueue('orders.alerts', 'orders', 'order.payment-failed');

• order.created → fulfillment + audit.
• order.cancelled → audit only.
• order.payment-failed → alerts only.

When to pick it: explicit separation of flows — alerts apart from audit, the main worker apart from monitoring.

Subscribe by a pattern — Topic

Routing is great for strict rules. But what if a service wants to subscribe to "all order events"? Or "everything from the EU region"?

A topic exchange lets you define subscriptions with patterns. Message keys are built with dots (order.created.eu), and in a subscription you can use:

• * — exactly one word,
• # — zero or more words.

await channel.assertExchange('events', 'topic', { durable: true });

const quorum = { durable: true, arguments: { 'x-queue-type': 'quorum' } };
await channel.assertQueue('audit.orders', quorum);
await channel.assertQueue('dashboard.eu', quorum);
await channel.assertQueue('alerts.critical', quorum);

await channel.bindQueue('audit.orders', 'events', 'order.#');
await channel.bindQueue('dashboard.eu', 'events', '*.*.eu');
await channel.bindQueue('alerts.critical', 'events', 'payment.failed.#');

A message with the key order.cancelled.eu will land in audit.orders (via order.#) and in dashboard.eu (via *.*.eu).

When to pick it: events with a hierarchical structure, when you need to subscribe flexibly without reworking the topology every time a new event type is added.

Request-response over a queue — RPC

Sometimes you need a synchronous response, but HTTP won't do: the service is behind NAT, has no public address, or you want load balancing across a pool of workers.

RPC over a queue: the client sends a request and waits for a response. The broker delivers the request to one of the workers, which replies to a separate reply queue. A correlation-id is used to match the request with the response.

In amqplib you assemble this by hand; RabbitMQ provides the fast amq.rabbitmq.reply-to pseudo-queue (direct reply-to):

// Client
const pending = new Map<string, (quote: PriceQuote) => void>();

await channel.consume('amq.rabbitmq.reply-to', (msg) => {
  if (!msg) return;
  pending.get(msg.properties.correlationId)?.(JSON.parse(msg.content.toString()));
  pending.delete(msg.properties.correlationId);
}, { noAck: true });

function quote(request: QuoteRequest): Promise<PriceQuote> {
  const correlationId = randomUUID();
  return new Promise((resolve) => {
    pending.set(correlationId, resolve);
    channel.publish('pricing.exchange', 'pricing.quote',
      Buffer.from(JSON.stringify(request)),
      { replyTo: 'amq.rabbitmq.reply-to', correlationId });
  });
}

// Server
await channel.consume('pricing.quote', (msg) => {
  if (!msg) return;
  const quote = computeQuote(JSON.parse(msg.content.toString()));
  channel.sendToQueue(msg.properties.replyTo, Buffer.from(JSON.stringify(quote)), {
    correlationId: msg.properties.correlationId,
  });
  channel.ack(msg);
});

The client subscribes to amq.rabbitmq.reply-to, sets reply-to and correlation-id, and matches responses with requests. The server publishes the response to the queue from the request's replyTo property.

When to pick it: you need a synchronous call, but HTTP doesn't work (NAT, firewall, no public address); you need to balance requests across a pool of workers.

When not to pick it: if HTTP/gRPC simply works — RPC over a broker is harder to debug and more expensive.

What to do about redelivery — Idempotent Consumer

AMQP guarantees at-least-once delivery: the same message may arrive twice. This happens when the broker didn't receive an acknowledgment for the processing (for example, because of a network issue) and re-sends the message.

A consumer must be able to handle repeats without breaking business logic.

Idempotency key in the database

The most reliable approach is to remember already-processed messages:

await channel.consume('payments', async (msg) => {
  if (!msg) return;
  const event: PaymentEvent = JSON.parse(msg.content.toString());
  await dataSource.transaction(async (tx) => {
    if (await tx.processedEvents.existsByIdempotencyKey(event.idempotencyKey)) {
      return; // already processed — just acknowledge receipt
    }
    await tx.processedEvents.save({ idempotencyKey: event.idempotencyKey });
    await tx.accounts.debit(event.accountId, event.amount);
  });
  channel.ack(msg);
});

A processed_events table with a unique index on idempotency_key. If two identical messages arrive at the same time, the database catches the duplicate through a unique constraint violation.

Checking the object's state

If the event moves an object into a new state, it's enough to check the current one:

async function onOrderConfirmed(event: OrderConfirmedEvent): Promise<void> {
  await dataSource.transaction(async (tx) => {
    const order = await tx.orders.findByIdOrFail(event.orderId);
    if (order.status === OrderStatus.CONFIRMED) {
      return; // already in the desired state
    }
    order.confirm();
    await tx.orders.save(order);
  });
}

This needs no separate table — the state is already stored in the business object.

Retry with a delay and a Dead Letter Queue

What if the consumer failed not because of a bug, but because an external service was temporarily unavailable? You want to try again, but not immediately.

With amqplib a delayed retry is assembled with x-message-ttl and a Dead Letter Exchange:

await channel.assertQueue('orders.retry', {
  durable: true,
  arguments: {
    'x-queue-type': 'quorum',
    'x-message-ttl': 30_000, // wait 30 seconds
    'x-dead-letter-exchange': 'orders',
    'x-dead-letter-routing-key': 'order.created',
  },
});

The flow: the consumer rejects the message → it lands in the retry queue → after 30 seconds, once the TTL expires, it goes back through the DLX into the main queue → a new attempt.

The number of attempts is counted via the x-death.count header — you have to check it manually; there is no built-in limiter.

Messages that couldn't be processed after all attempts go to a Dead Letter Queue (DLQ) — a separate queue for manual inspection or alerts.

Guaranteed publishing — Outbox

Here's a common task: save an order to the database and publish an event — atomically. If you save first and publish afterward, the service may crash between the two operations. The event is lost.

The Outbox pattern: the event is saved in the same transaction as the business data. A separate process reads the table and publishes to AMQP.

@Injectable()
export class OrderService {
  async confirm(orderId: OrderId): Promise<void> {
    await this.dataSource.transaction(async (tx) => {
      const order = await tx.orders.findByIdOrFail(orderId);
      order.confirm();
      await tx.orders.save(order);
      await tx.outbox.save({
        id: randomUUID(),
        routingKey: 'order.confirmed',
        exchange: 'orders',
        payload: JSON.stringify(new OrderConfirmedEvent(orderId)),
      });
    });
  }

  @Interval(500)
  async publishOutbox(): Promise<void> {
    const batch = await this.outboxRepo.fetchUnpublished(100);
    for (const event of batch) {
      this.channel.publish(event.exchange, event.routingKey, Buffer.from(event.payload));
      await this.outboxRepo.markPublished(event.id);
    }
  }
}

Either both changes are committed, or neither is. Duplicates are possible (the publish succeeded, but marking it as sent didn't finish in time) — which is why the receiver still has to be idempotent.

Selection cheat sheet

In short

• Work Queue — one queue, several workers, each message goes to exactly one. For background tasks.
• Publish/Subscribe — a fanout exchange copies the message into all bound queues. For broadcast events.
• Routing — a direct exchange looks at the routing key. For precise separation of flows.
• Topic — like routing, but with * and # patterns. For hierarchical events with flexible subscription.
• RPC over a queue — request-response through the broker with reply-to and correlation-id. For calls without HTTP.
• Idempotent Consumer — at-least-once means possible duplicates. Protection: an idempotency key in the DB or a check of the object's state.
• Delayed Retry — TTL + DLX: the message is "parked" for a while, then comes back.
• Outbox — the event is saved in the same transaction as the data. Atomicity without a two-phase commit.

What to read next

• The AMQP protocol — the exchange/binding/queue model from the inside.
• Spring AMQP — configuration, RabbitTemplate, annotations.
• RabbitMQ in production — Quorum Queues, clustering, monitoring.
• AMQP vs Kafka — which broker to pick and when.