RabbitMQ Node.js/TypeScript Skill

Keep a single connection per process and create channels for logical units of work. Channels are lightweight and can be reopened after errors.

Publishing messages

async function publish(channel: Channel, queue: string, msg: string) {
  await channel.assertQueue(queue, { durable: true });
  channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });
}

// exchange example
await channel.assertExchange('logs', 'fanout', { durable: false });
channel.publish('logs', '', Buffer.from('hello world'));

• Use assertQueue/assertExchange to ensure the topology exists.
• Mark persistent messages and durable queues if you need them to survive broker restarts.

Consuming messages

async function consume(channel: Channel, queue: string, handler: (msg: string) => void) {
  await channel.assertQueue(queue, { durable: true });
  await channel.consume(queue, msg => {
    if (msg) {
      handler(msg.content.toString());
      channel.ack(msg);
    }
  });
}

• Always ack or nack messages; otherwise they will be re-queued when the consumer connection closes.
• You can use channel.prefetch(1) to process one message at a time and avoid overloading consumers.

Exchange types and routing

RabbitMQ supports several exchange types:

Binding queues to exchanges allows flexible routing; call channel.bindQueue(queue, exchange, routingKey).

TypeScript tips

• The amqplib module exports interfaces such as Connection, Channel, and Message you can import directly.
• Create wrapper types for your message payloads and validate/serialize with a library like zod or io-ts before publishing.

interface TaskPayload { taskId: string; data: any; }
channel.sendToQueue('tasks', Buffer.from(JSON.stringify(payload)));

Common pitfalls

• Forgetting to close connections - call conn.close() on shutdown to flush any outstanding confirms.
• Blocking the event loop - don't perform heavy CPU work in the consumer callback; offload to a worker or thread pool.
• Unacknowledged messages - if your process crashes without acking, messages will be redelivered; design idempotent handlers.
• Reconnection logic - handle error/close events on the connection and attempt to reconnect with backoff.
• Missing await - many methods return promises; neglecting them can lead to lost messages or unexpected ordering.

Advanced features

• Confirm channels: conn.createConfirmChannel() returns a channel that provides publisher confirms (channel.waitForConfirms()). Use channel.waitForConfirms() or the callback form of sendToQueue to know when the broker has accepted a message.
• RPC pattern: use a combination of private reply queues and the replyTo/correlationId properties. Keep a single reply queue per client and correlate responses.
• Dead-letter exchanges: configure a queue with deadLetterExchange to handle unprocessable messages, optionally with TTL/expiration for retries.
• Message TTL / expirations: set messageTtl or expiration on queues or properties to auto-expire stale messages.
• QoS and prefetch: use channel.prefetch(n) to control how many unacked messages are sent to each consumer; useful for batching and flow control.
• Clustered brokers / high availability: configure mirrored queues or use quorum queues (RabbitMQ 3.8+) for resilience. The client connection URI can include multiple hosts for failover, e.g. amqp://host1,host2.

Configuration & environment variables

Many applications keep connection parameters in environment variables:

RABBITMQ_URL=amqp://user:pass@rabbitmq:5672/vhost
RABBITMQ_PREFETCH=5          # optional prefetch count
RABBITMQ_HEARTBEAT=30        # seconds

Pass them to amqp.connect(process.env.RABBITMQ_URL) and call channel.prefetch(+process.env.RABBITMQ_PREFETCH || 0).

Reconnection strategy

Handle error and close events on both connection and channel objects. A simple backoff loop looks like:

async function start() {
  while (true) {
    try {
      const { conn, channel } = await connect();
      await consume(channel, 'tasks', handler);
      break; // consumed successfully
    } catch (err) {
      console.error('rabbitmq error, retrying in 5s', err);
      await new Promise(r => setTimeout(r, 5000));
    }
  }
}

Libraries like amqp-connection-manager wrap this pattern and automatically reopen channels when the broker restarts.

Example usage

(async () => {
  const { conn, channel } = await connect();
  await publish(channel, 'hello', 'world');
  await consume(channel, 'hello', console.log);
  // teardown...
  await channel.close();
  await conn.close();
})();