Connections & Channels

The number-one production mistake

More production RabbitMQ incidents come from connection misuse than from anything else. A TCP connection with its TLS handshake and AMQP negotiation is expensive to open. Code that opens a connection per message — or per request — floods the broker with churn, exhausts file descriptors, and falls over under load.

The rule from the foundations module, restated for production:

One long-lived connection per process. Many channels over it — one per worker or concurrent task. Channels are not thread-safe.

flowchart LR
  app["App process"] --> conn["1 long-lived
connection"]
  conn --> ch1["Channel → worker 1"]
  conn --> ch2["Channel → worker 2"]
  conn --> ch3["Channel → worker 3"]
  ch1 --> broker["RabbitMQ"]
  ch2 --> broker
  ch3 --> broker

The correct shape: one connection, many channels, one channel per worker

Heartbeats keep connections honest

A connection can look alive at the TCP level while the peer is actually gone (a crashed node, a silent network drop). Heartbeats are periodic frames both sides exchange; if several are missed, the connection is considered dead and torn down. This lets a client detect a broken broker in seconds instead of waiting for a TCP timeout that can take minutes. Keep the default heartbeat on — turning it off is how “zombie” connections happen.

Automatic recovery

Networks blip. A robust client does not crash when a connection drops — it reconnects, then re-declares its topology (queues, exchanges, bindings) and re-establishes its consumers. Most clients offer this, but you must design for it: a redelivered message may arrive after recovery, so your consumers still need to be idempotent.

import amqp from 'amqplib';

// amqplib has no built-in recovery — wrap connect with retry.
async function connectWithRetry(url, attempt = 0) {
  try {
    const conn = await amqp.connect(url, { heartbeat: 30 });
    conn.on('error', (e) => console.error('conn error', e.message));
    conn.on('close', () => setTimeout(() => connectWithRetry(url), 2000));
    return conn;
  } catch (e) {
    const delay = Math.min(2 ** attempt * 500, 30000);
    console.warn(`connect failed, retry in ${delay}ms`);
    await new Promise((r) => setTimeout(r, delay));
    return connectWithRetry(url, attempt + 1);
  }
}

import pika

# pika's SelectConnection supports recovery; BlockingConnection needs a retry loop.
params = pika.ConnectionParameters(
    host="localhost",
    heartbeat=30,
    connection_attempts=5,
    retry_delay=2,          # seconds between attempts
    blocked_connection_timeout=300,
)
conn = pika.BlockingConnection(params)
channel = conn.channel()   # re-declare topology after any reconnect

// amqp091-go has no auto-recovery — loop on the connection's NotifyClose.
for {
    conn, err := amqp.DialConfig("amqp://guest:guest@localhost:5672/",
        amqp.Config{Heartbeat: 30 * time.Second})
    if err != nil {
        time.Sleep(2 * time.Second)
        continue
    }
    closed := conn.NotifyClose(make(chan *amqp.Error, 1))
    runWorkers(conn)     // declare topology + consumers here
    <-closed             // block until the connection drops, then reconnect
    log.Println("connection lost, reconnecting...")
}

A short production checklist

One connection per process, reused for the app’s lifetime.
One channel per worker/goroutine/thread — never share a channel across threads.
Separate the publisher channel from consumer channels (a channel error, like publishing to a missing exchange, closes the whole channel).
Keep heartbeats on so dead connections are detected quickly.
Reconnect with backoff and re-declare topology + consumers on recovery.