Pola Integrasi: Work Queue, Pub/Sub, RPC, dan Outbox — Belajar RabbitMQ

Pola 1 — Work Queue (Competing Consumers)

Pola paling umum: pekerjaan berat dipindahkan dari jalur request agar pengguna tidak menunggu.

work queuetext

HTTP request ──> API ──publish──> [ queue ] ──> worker 1
                  │                     ├──────> worker 2
                  └── balas 202          └──────> worker 3
                      "sedang diproses"

Menambah kapasitas = menambah worker pada queue yang sama.

💡 Kembalikan identitas pekerjaan, bukan hasilnya

API membalas 202 Accepted beserta job_id. Klien memantau statusnya lewat polling atau webhook. Jangan membuat HTTP request menunggu selesainya pekerjaan asinkron — itu menghapus seluruh manfaat pemisahannya.

Pola 2 — Publish/Subscribe

Satu peristiwa, banyak pihak yang bereaksi secara independen. Kunci pola ini: producer tidak tahu siapa yang mendengarkan.

pub/subtext

                          ┌──> [ queue: kirim-email ]   ──> layanan email
publish "pesanan.dibuat" ─┼──> [ queue: kurangi-stok ]  ──> layanan gudang
     ke exchange topic    ├──> [ queue: catat-audit ]   ──> layanan audit
                          └──> [ queue: analitik ]      ──> data pipeline

Menambah reaksi baru = menambah queue + binding.
Kode producer TIDAK berubah sama sekali.

⚠️ Event vs Command — bedakan sejak awal

text

EVENT   : "pesanan.dibuat"       (lampau, fakta)
  → boleh punya nol atau banyak pendengar
  → producer tidak peduli siapa yang bereaksi
  → exchange topic/fanout

COMMAND : "kirim.email"          (perintah, imperatif)
  → tepat satu pemroses
  → pengirim mengharapkan sesuatu dilakukan
  → exchange direct, satu queue

Mencampur keduanya menghasilkan sistem yang membingungkan: orang menambahkan konsumer kedua pada sebuah command dan tiba-tiba email terkirim dua kali.

Pola 3 — RPC (Request/Reply)

Kadang Anda memang butuh jawaban. RabbitMQ mendukungnya lewat reply_to dan correlation_id.

klien RPCjavascript

const { randomUUID } = require('crypto');

class KlienRPC {
  async siapkan(ch) {
    this.ch = ch;
    this.menunggu = new Map();

    // pseudo-queue balasan: tanpa deklarasi, tanpa biaya
    await ch.consume('amq.rabbitmq.reply-to', (msg) => {
      const id = msg.properties.correlationId;
      const p = this.menunggu.get(id);
      if (!p) return;                         // balasan telat, abaikan
      this.menunggu.delete(id);
      clearTimeout(p.timer);
      p.resolve(JSON.parse(msg.content.toString()));
    }, { noAck: true });                      // direct reply-to WAJIB noAck
  }

  panggil(rk, payload, timeoutMs = 5000) {
    const id = randomUUID();
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.menunggu.delete(id);
        reject(new Error('RPC timeout: ' + rk));
      }, timeoutMs);

      this.menunggu.set(id, { resolve, timer });

      this.ch.publish('rpc', rk, Buffer.from(JSON.stringify(payload)), {
        correlationId: id,
        replyTo: 'amq.rabbitmq.reply-to',
        expiration: String(timeoutMs)         // pesan mati bila tak terlayani
      });
    });
  }
}

server RPCjavascript

await ch.prefetch(10);
await ch.consume('rpc.hitung-ongkir', async (msg) => {
  let hasil;
  try {
    hasil = { ok: true, data: await hitungOngkir(JSON.parse(msg.content)) };
  } catch (e) {
    hasil = { ok: false, error: e.message };  // balas error, jangan diam
  }
  ch.publish('', msg.properties.replyTo, Buffer.from(JSON.stringify(hasil)), {
    correlationId: msg.properties.correlationId
  });
  ch.ack(msg);
}, { noAck: false });

⛔ RPC lewat broker sering merupakan pilihan yang salah

Bila Anda menunggu jawaban secara sinkron, Anda sudah kehilangan hampir semua manfaat pesan asinkron — dan menambah satu komponen yang bisa gagal di tengah jalur.

text

RPC lewat RabbitMQ masuk akal bila:
  ✓ butuh load balancing otomatis ke banyak pemroses
  ✓ pemroses berada di jaringan yang tidak bisa dihubungi langsung
  ✓ sudah ada infrastruktur RabbitMQ dan HTTP tidak tersedia

Gunakan HTTP/gRPC bila:
  ✓ hubungan sederhana satu ke satu
  ✓ butuh latensi serendah mungkin
  ✓ butuh streaming atau status kode yang kaya

Selalu pasang timeout dan expiration pada pesan RPC. Tanpa keduanya, permintaan yang tidak terlayani menumpuk selamanya dan klien menggantung.

Pola 4 — Transactional Outbox

Ini pola terpenting yang paling sering diabaikan. Masalahnya: Anda tidak bisa melakukan commit basis data dan publish ke broker secara atomik.

masalah dual writetext

❌ Urutan A:
   BEGIN; INSERT pesanan; COMMIT;
   publish("pesanan.dibuat")     ← proses mati di sini
   → pesanan ada di DB, tetapi TIDAK ADA yang tahu. Stok tak berkurang.

❌ Urutan B:
   publish("pesanan.dibuat")
   BEGIN; INSERT pesanan; COMMIT;   ← gagal di sini
   → email terkirim untuk pesanan yang tidak pernah ada.

solusi: tabel outboxsql

CREATE TABLE outbox (
  id             BIGSERIAL PRIMARY KEY,
  agregat_tipe   TEXT        NOT NULL,
  agregat_id     TEXT        NOT NULL,
  routing_key    TEXT        NOT NULL,
  payload        JSONB       NOT NULL,
  dibuat_pada    TIMESTAMPTZ NOT NULL DEFAULT now(),
  dikirim_pada   TIMESTAMPTZ
);

CREATE INDEX outbox_belum_terkirim
  ON outbox (id) WHERE dikirim_pada IS NULL;

penulisan atomikjavascript

// 1. satu transaksi: data bisnis DAN niat mengirim pesan
await db.transaction(async (trx) => {
  const pesanan = await trx('pesanan').insert(dataPesanan).returning('*');
  await trx('outbox').insert({
    agregat_tipe: 'pesanan',
    agregat_id:   pesanan.id,
    routing_key:  'pesanan.jakarta.dibuat',
    payload:      JSON.stringify(pesanan)
  });
});
// keduanya berhasil, atau keduanya batal. Tidak ada keadaan setengah.

// 2. proses terpisah memindahkan outbox ke RabbitMQ
async function pengirimOutbox() {
  const baris = await db('outbox')
    .whereNull('dikirim_pada')
    .orderBy('id')
    .limit(200)
    .forUpdate().skipLocked();        // aman untuk banyak instance

  for (const r of baris) {
    await publishDenganConfirm('pesanan', r.routing_key, r.payload, {
      messageId: String(r.id)          // kunci deduplikasi konsumer
    });
    await db('outbox').where({ id: r.id }).update({ dikirim_pada: db.fn.now() });
  }
}

🎯 Konsekuensi yang harus diterima

Outbox memberi jaminan at-least-once, bukan exactly-once: bila proses mati setelah publish berhasil tetapi sebelum menandai dikirim_pada, pesan itu akan dikirim dua kali.

Itu bukan kelemahan pola ini, melainkan sifat sistem terdistribusi. Karena itu konsumer wajib idempoten — dibahas di bab berikutnya. Dengan message_id dari outbox.id, deduplikasi menjadi sangat sederhana.

Saga: Transaksi Lintas Layanan

Ketika satu proses bisnis melibatkan beberapa layanan, tidak ada transaksi global. Yang dipakai adalah rangkaian langkah, masing-masing dengan langkah pembatalnya.

saga koreografitext

pesanan.dibuat
   └──> layanan stok: kurangi stok
          ├── berhasil ──> stok.dikurangi
          │                   └──> layanan bayar: tagih
          │                          ├── berhasil ──> pembayaran.berhasil
          │                          │                  └──> pesanan.dikonfirmasi
          │                          └── gagal ────> pembayaran.gagal
          │                                            └──> stok.kembalikan (kompensasi)
          │                                                   └──> pesanan.dibatalkan
          └── gagal ─────> stok.habis
                             └──> pesanan.dibatalkan

Setiap langkah wajib punya kompensasi yang idempoten dan aman dijalankan berulang. Untuk alur dengan lebih dari empat langkah, koreografi menjadi sulit dilacak — pertimbangkan orkestrasi dengan satu koordinator yang menyimpan state saga di basis data.