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.

producer → [single queue] → consumer 1
                          → consumer 2
                          → consumer 3

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.

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

Task	Pattern	Exchange type
Distribute load across workers	Work Queue	direct (default)
Broadcast events to all services	Publish/Subscribe	fanout
Different events to different queues	Routing	direct
Pattern subscription to hierarchical events	Topic	topic
Synchronous call over a queue	RPC	direct + reply-to
Protection against redelivery	Idempotent Consumer	any
Retry with a delay	Delayed Retry	direct + DLX
Atomic publishing together with a DB write	Outbox	direct

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.