amqplib

Overview

amqplib speaks AMQP 0-9-1 from Node.js. It is only the client: the broker, exchange types, policies, clustering and operations are covered by the rabbitmq skill. This skill is about the Node API and its pitfalls.

Version 2.x (2.2.0 in September 2026) keeps the 0.x/1.x API. What changed since the old tutorials:

• Types are bundled (since 1.2.0). Do not install @types/amqplib; the DefinitelyTyped package stopped at 0.10.x.
• Node.js 18 or newer (since 1.0.0), and 0.10.6 or later is required for RabbitMQ 4.1+ (frame size).
• Connection recovery is built in (since 1.1.0): pass recovery to connect and the connection reconnects with backoff.
• heartbeat: 0 now disables heartbeats (2.0.0). Before, it meant "use the server's value". Omit the option for the server default.
• The promise API is the default import (amqplib); the callback API lives in amqplib/callback_api.

Instructions

Install

npm install amqplib
docker run -d --name rabbit -p 127.0.0.1:5672:5672 -p 127.0.0.1:15672:15672 rabbitmq:4-management

Management UI: http://localhost:15672. Keep the broker URL in an environment variable such as RABBITMQ_URL (amqp://orders_app:<password>@localhost:5672/shop, TLS: amqps://). Do not use the default guest account outside the machine running the broker.

Connect once, open channels

A connection is one TCP socket; keep one per process. A channel is a lightweight session on it. Use one channel for publishing and a separate one for consuming, because a consumer that is blocked by prefetch should not stall publishes.

import amqp, { type ChannelModel, type ConsumeMessage } from "amqplib";

const connection = await amqp.connect(process.env.RABBITMQ_URL!);
connection.on("error", (err) => console.error("connection error", err.message));
connection.on("close", () => console.error("connection closed"));

A channel closes on any channel-level error (publishing to a missing exchange, acking twice, redeclaring a queue with different arguments). Attach channel.on("error", ...) and channel.on("close", ...) and open a new channel when it happens.

Declare topology with a dead-letter queue

const ch = await connection.createChannel();
await ch.assertExchange("orders.dlx", "fanout", { durable: true });
await ch.assertQueue("orders.failed", { durable: true });
await ch.bindQueue("orders.failed", "orders.dlx", "");
await ch.assertQueue("orders", {
  durable: true,
  arguments: { "x-dead-letter-exchange": "orders.dlx", "x-message-ttl": 86_400_000 },
});

assertQueue fails (and closes the channel) if the queue exists with different arguments. Change arguments by creating a new queue name or by using a RabbitMQ policy.

Publish with confirms

A plain channel's sendToQueue returns before the broker has the message. Use a confirm channel when losing a message matters:

const pub = await connection.createConfirmChannel();

async function publishOrder(order: { orderId: string; totalCents: number }) {
  const ok = pub.sendToQueue("orders", Buffer.from(JSON.stringify(order)), {
    persistent: true,                 // written to disk; the queue must be durable too
    contentType: "application/json",
    messageId: order.orderId,
  });
  if (!ok) await new Promise((resolve) => pub.once("drain", resolve)); // write buffer is full
  await pub.waitForConfirms();        // rejects if the broker nacks any outstanding publish
}

publish(exchange, routingKey, content, options) is the exchange variant. A message published to an exchange with no matching queue is dropped silently unless you set mandatory: true and listen to the channel's return event.

Consume with manual acknowledgment

const sub = await connection.createChannel();
await sub.prefetch(10);               // at most 10 unacknowledged messages per consumer

await sub.consume("orders", async (msg) => {
  if (msg === null) return;           // the broker cancelled the consumer (queue deleted, node down)
  try {
    await handleOrder(JSON.parse(msg.content.toString()));
    sub.ack(msg);
  } catch (err) {
    // requeue once, then dead-letter; msg.fields.redelivered tells which attempt this is
    sub.nack(msg, false, !msg.fields.redelivered);
  }
});

Never pass noAck: true for work that must not be lost. Ack and nack each message exactly once, on the channel that delivered it. An unhandled exception inside the callback leaves the message unacknowledged until the channel closes.

Reconnect automatically

const conn = await amqp.connect(process.env.RABBITMQ_URL!, {
  recovery: {
    initialDelay: 200,                // ms
    maxDelay: 30_000,
    maxRetries: Infinity,
    initialMaxRetries: 5,             // fail fast if the broker is down at startup
    async setup(model: ChannelModel) {   // runs after every (re)connect; the annotation is needed for TypeScript
      const ch = await model.createChannel();
      await ch.prefetch(10);
      await ch.assertQueue("orders", {   // same arguments as the first declaration, or the broker answers 406
        durable: true,
        arguments: { "x-dead-letter-exchange": "orders.dlx", "x-message-ttl": 86_400_000 },
      });
      await ch.consume("orders", (msg: ConsumeMessage | null) => { if (msg) ch.ack(msg); });
    },
  },
});
conn.on("disconnect", (err) => console.warn("lost broker:", err.message));
conn.on("reconnect-scheduled", ({ attempt, delay }) => console.log(`retry ${attempt} in ${delay} ms`));
conn.on("reconnect-failed", (err) => { console.error(err); process.exit(1); });

Recovery re-establishes the connection only. Channels, queues, bindings and consumers created before the drop are gone, so put every declaration and consume call in setup. Options: waitForConnect: false returns the handle immediately (await conn.waitForConnect() later), calculateDelay(attempt) replaces the built-in backoff (2.2.0; your function must enforce its own cap).

Examples

Example 1: Send order jobs to a worker

Request: "Our checkout service in Node should hand off confirmation emails to a worker, and nothing may be lost if the worker restarts."

Producer in checkout.ts, using the confirm channel from above:

await publishOrder({ orderId: "ORD-20481", totalCents: 12990 });

Worker in mail-worker.ts: the consumer from the previous section with handleOrder sending the email. Start it, then run docker exec rabbit rabbitmqctl list_queues name messages_ready messages_unacknowledged:

orders    0    0

Stop the worker, publish three more orders, and the same command shows orders 3 0. Restart the worker and it drains them; a message whose handler throws twice lands in orders.failed.

Example 2: Topic events for several services

Request: "Billing wants every order.* event, analytics wants everything."

const ch = await connection.createChannel();
await ch.assertExchange("events", "topic", { durable: true });

async function subscribe(queueName: string, pattern: string, onEvent: (key: string, body: unknown) => Promise<void>) {
  await ch.assertQueue(queueName, { durable: true });   // named and durable: events wait while the service is down
  await ch.bindQueue(queueName, "events", pattern);
  await ch.consume(queueName, async (msg) => {
    if (!msg) return;
    await onEvent(msg.fields.routingKey, JSON.parse(msg.content.toString()));
    ch.ack(msg);
  });
}

await subscribe("billing.orders", "order.*", async (key, body) => console.log(key, body));
await subscribe("analytics.all", "#", async (key, body) => console.log(key, body));

ch.publish("events", "order.created", Buffer.from(JSON.stringify({ orderId: "ORD-20482", items: 3 })), { persistent: true });

Both queues get order.created; a later user.signup reaches only analytics.all.

Guidelines

• Messages are Buffers. Serialize and parse yourself, and set contentType so other consumers know the format.
• Durable queue plus persistent: true are both needed to survive a broker restart. Persistence alone is not delivery: use a confirm channel.
• Delivery is at least once. Make handlers idempotent, for example by recording messageId.
• Exclusive, auto-delete queues (assertQueue("", { exclusive: true })) vanish with the connection; use them only for throwaway subscribers.
• Do not open a connection per message. Do not share one channel between publishers and long-running consumers.
• On shutdown call await channel.close() then await connection.close() so in-flight acks are sent.
• Quorum queues (arguments: { "x-queue-type": "quorum" }) are the recommended replicated type on RabbitMQ 4; classic mirrored queues are gone.
• Broker setup, policies, retries with delay exchanges and monitoring belong to the rabbitmq skill.