Klien Produksi: Reconnect, Heartbeat, dan Graceful Shutdown — Belajar RabbitMQ

Koneksi Akan Putus — Itu Normal

Broker di-upgrade, load balancer memutus koneksi menganggur, jaringan berkedip, node di-failover. Aplikasi produksi harus menganggap putusnya koneksi sebagai peristiwa rutin, bukan kejadian luar biasa.

Heartbeat: Mendeteksi Koneksi yang Mati Diam-diam

kenapa heartbeat perlutext

Tanpa heartbeat:
  Kabel dicabut / firewall membuang state NAT
  → soket TCP tidak tahu apa-apa
  → aplikasi mengira masih terhubung
  → publish menghilang ke ketiadaan selama BERJAM-JAM
  → sampai TCP keepalive OS akhirnya menyerah (default 2 jam!)

Dengan heartbeat = 60:
  Kedua sisi bertukar frame tiap 30 detik (setengah interval).
  Dua kali terlewat → koneksi dinyatakan mati dalam ± 60 detik.

⚠️ Heartbeat 0 berarti mematikan deteksi

Beberapa tutorial menyarankan heartbeat=0 untuk menghindari "connection reset" saat debugging. Di produksi ini berbahaya: koneksi mati tidak akan pernah terdeteksi, dan pesan hilang tanpa jejak.

Nilai 60 detik cocok untuk hampir semua kasus. Untuk konsumer dengan pemrosesan sangat lama, pastikan handler tidak memblokir event loop sehingga frame heartbeat tetap terkirim.

Reconnect dengan Exponential Backoff

lapisan koneksi tahan bantingjavascript

const amqp = require('amqplib');

class Koneksi {
  constructor(url, nama) {
    this.url = url;
    this.nama = nama;
    this.conn = null;
    this.ch = null;
    this.tutupSengaja = false;
    this.percobaan = 0;
    this.pendengar = [];          // dipanggil ulang setiap reconnect
  }

  async sambung() {
    try {
      this.conn = await amqp.connect(this.url, {
        heartbeat: 60,
        clientProperties: { connection_name: this.nama }
      });

      this.conn.on('error', (e) => console.error('conn error:', e.message));
      this.conn.on('close', () => {
        if (this.tutupSengaja) return;
        console.warn('koneksi tertutup, menyambung ulang…');
        this.jadwalkan();
      });

      this.ch = await this.conn.createConfirmChannel();
      this.ch.on('error', (e) => console.error('channel error:', e.message));

      this.percobaan = 0;
      // WAJIB: topologi dan konsumer dipasang ULANG setiap reconnect
      for (const pasang of this.pendengar) await pasang(this.ch);
      console.log('tersambung.');
    } catch (e) {
      console.error('gagal menyambung:', e.message);
      this.jadwalkan();
    }
  }

  jadwalkan() {
    this.ch = null;
    const dasar = Math.min(1000 * Math.pow(2, this.percobaan++), 30000);
    const jitter = Math.random() * dasar * 0.3;   // cegah thundering herd
    setTimeout(() => this.sambung(), dasar + jitter);
  }

  daftar(pasang) { this.pendengar.push(pasang); }

  async tutup() {
    this.tutupSengaja = true;
    if (this.ch) await this.ch.close().catch(() => {});
    if (this.conn) await this.conn.close().catch(() => {});
  }
}

⛔ Jangan lupa memasang ulang konsumer

Kesalahan reconnect paling umum: koneksi tersambung kembali, log menunjukkan "tersambung", tetapi basic.consume tidak pernah dipanggil ulang.

Akibatnya aplikasi tampak sehat sementara antrean menumpuk diam-diam. Deklarasi exchange, queue, binding, prefetch, dan consume semuanya melekat pada channel — dan channel ikut hilang bersama koneksi.

Jitter Bukan Detail Kosmetik

Bila 200 pod kehilangan koneksi bersamaan saat broker restart, backoff tanpa jitter membuat semuanya mencoba menyambung pada detik yang sama persis. Broker yang baru bangun langsung dihantam 200 handshake serentak dan bisa jatuh lagi — lalu siklusnya berulang.

Pemakaian Channel yang Benar

aturan channeltext

✅ Satu connection per proses aplikasi
✅ Satu channel per thread / goroutine / worker
✅ Channel publisher terpisah dari channel consumer
✅ Channel dibuat ulang bila ditutup broker

❌ Satu channel dipakai bersama beberapa thread
❌ Channel baru untuk setiap pesan (mahal, ± 1 round-trip)
❌ Satu connection untuk seluruh cluster aplikasi

💡 Channel bisa ditutup sepihak oleh broker

Kesalahan tingkat channel — mendeklarasikan queue yang sudah ada dengan argumen berbeda, mem-publish ke exchange yang tidak ada, meng-ack delivery tag yang tidak dikenal — membuat broker menutup channel, bukan mengembalikan error biasa.

Karena itu selalu pasang handler channel.on('error') dan siapkan pembuatan ulang channel. Koneksinya sendiri biasanya tetap sehat.

Graceful Shutdown

shutdown yang benarjavascript

let sedangProses = 0;
let berhenti = false;

async function matikan() {
  berhenti = true;

  // 1. berhenti menerima pesan baru — pesan yang belum diambil
  //    tetap aman di queue
  await ch.cancel(consumerTag);

  // 2. tunggu pekerjaan yang sedang berjalan
  const batas = Date.now() + 25000;
  while (sedangProses > 0 && Date.now() < batas) {
    await new Promise(r => setTimeout(r, 200));
  }
  if (sedangProses > 0) {
    console.warn(sedangProses + ' pekerjaan belum selesai; akan dikirim ulang.');
  }

  // 3. tutup dengan rapi
  await ch.close().catch(() => {});
  await conn.close().catch(() => {});
  process.exit(0);
}

process.on('SIGTERM', matikan);
process.on('SIGINT', matikan);

k8s harus memberi cukup waktuyaml

spec:
  terminationGracePeriodSeconds: 40    # > 25 detik di atas + margin
  containers:
    - name: worker
      lifecycle:
        preStop:
          exec:
            command: ["sleep", "5"]    # beri jeda agar endpoint dicabut dulu

Klien yang Dianjurkan

Bahasa	Pustaka	Catatan
Node.js	`amqplib`	Reconnect harus ditulis sendiri
Go	`amqp091-go`	Resmi; `NotifyClose` untuk deteksi putus
Java / Kotlin	`amqp-client` / Spring AMQP	Spring AMQP menangani reconnect otomatis
Python	`pika` (async) / `aio-pika`	`aio-pika` punya reconnect bawaan
.NET	`RabbitMQ.Client`	`AutomaticRecoveryEnabled` tersedia

Fitur "automatic recovery" pada beberapa klien memang memasang ulang topologi, tetapi perilakunya berbeda-beda dalam hal konsumer dan pesan yang belum di-ack. Uji sendiri dengan mematikan broker di tengah beban — jangan percaya begitu saja.