Bab 1

RabbitMQ

Belajar RabbitMQ dengan React, Node.js, dan MongoDB

Panduan praktis dari nol sampai jalan: Frontend → Queue → Consumer → MongoDB → Frontend

Daftar Isi#

1. Pendahuluan dan Arsitektur
2. Konsep Dasar RabbitMQ
3. Menyiapkan Environment dengan Docker
4. Mengenal Management Dashboard
5. Struktur Project
6. Backend: Helper Koneksi RabbitMQ
7. Backend: API (Producer)
   1. Backend: Consumer Worker
8. Frontend React
9. Menjalankan dan Menguji
10. Konsep Lanjutan: Ack, Retry, dan Dead Letter Queue
11. Exchange: Direct, Fanout, Topic
12. Best Practice
13. Troubleshooting
14. Migrasi ke FunctionGraph
15. Latihan Mandiri
16. Penutup dan Cheat Sheet

Bab 1: Pendahuluan dan Arsitektur#

1.1 Kenapa pakai message queue?#

Bayangkan frontend mengirim 1.000 data sekaligus. Kalau backend langsung menulis ke database satu per satu di dalam request HTTP:

• User menunggu lama (request lambat).
• Kalau database lambat atau mati, request gagal dan data hilang.
• Lonjakan traffic bisa menumbangkan backend.

Dengan message queue seperti RabbitMQ:

• API cukup menaruh pesan di antrean, lalu langsung membalas 202 Accepted.
• Consumer mengambil pesan dengan kecepatannya sendiri.
• Kalau consumer mati, pesan tetap aman di queue dan diproses saat consumer hidup lagi.

Pola ini disebut asynchronous processing atau decoupling.

1.2 Arsitektur yang akan kita bangun#

Ini versi lokal dari Test Case 4 di diagram kamu:

┌──────────────┐   POST /api/messages   ┌──────────────────┐
│  React App   │ ─────────────────────► │  API (Express)   │
│  (Vite)      │                        │  = "trigger      │
│              │ ◄───────────────────── │    backend"      │
│              │   GET /api/messages    └───────┬──────────┘
└──────────────┘                                │ publish
                                                ▼
                                        ┌──────────────┐
                                        │  RabbitMQ    │
                                        │ messages.queue│
                                        └──────┬───────┘
                                               │ consume
                                               ▼
                                        ┌──────────────────┐
                                        │ Consumer Worker  │
                                        │ = "backend       │
                                        │   consumer"      │
                                        └──────┬───────────┘
                                               │ write
                                               ▼
                                        ┌──────────────┐
                                        │   MongoDB    │
                                        └──────────────┘

1.3 Pemetaan diagram asli ke versi lokal#

1.4 Prasyarat#

• Node.js 20 LTS atau lebih baru
• Docker dan Docker Compose
• Editor kode (VS Code)
• Dasar JavaScript dan React (useState, useEffect)

Cek versi:

node -v
docker -v
docker compose version

Bab 2: Konsep Dasar RabbitMQ#

RabbitMQ adalah message broker yang mengimplementasikan protokol AMQP 0-9-1. Pahami istilah berikut sebelum menulis kode.

2.1 Komponen utama#

2.2 Alur pesan#

Producer → Exchange → (Binding) → Queue → Consumer

Producer tidak pernah mengirim langsung ke queue. Selalu lewat exchange. Kalau kita pakai default exchange (nama kosong ""), RabbitMQ otomatis meneruskan pesan ke queue yang namanya sama dengan routing key. Itulah yang dipakai sendToQueue() di ebook ini. Cara ini paling sederhana untuk belajar.

2.3 Acknowledgement (ack)#

Setelah consumer menerima pesan, RabbitMQ menunggu konfirmasi:

• ack: “Sudah beres, hapus dari queue.”
• nack / reject: “Gagal.” Pesan bisa dikembalikan ke queue atau dibuang ke dead letter queue.
• Consumer mati sebelum ack: RabbitMQ otomatis mengembalikan pesan ke queue (redelivery).

Ini yang membuat RabbitMQ aman: pesan tidak hilang hanya karena consumer crash.

2.4 Durability dan persistence#

Supaya pesan selamat saat RabbitMQ restart, dua hal harus aktif:

1. Queue dideklarasikan durable: true.
2. Pesan dikirim dengan persistent: true.

Kalau salah satunya tidak aktif, pesan bisa hilang saat broker restart.

2.5 Prefetch#

prefetch(n) membatasi berapa pesan yang boleh “menggantung” (belum di-ack) di satu consumer. Tanpa ini, satu consumer bisa menimbun ribuan pesan sementara consumer lain menganggur.

Bab 3: Menyiapkan Environment dengan Docker#

Buat folder project:

mkdir rabbitmq-react-lab
cd rabbitmq-react-lab

Buat file docker-compose.yml:

services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: lab-rabbitmq
    ports:
      - "5672:5672"     # AMQP (dipakai aplikasi)
      - "15672:15672"   # Management dashboard (dibuka di browser)
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  mongo:
    image: mongo:7
    container_name: lab-mongo
    ports:
      - "27017:27017"
    volumes:
      - mongo_data:/data/db

volumes:
  rabbitmq_data:
  mongo_data:

Jalankan:

docker compose up -d
docker compose ps

Kedua container harus berstatus running (RabbitMQ akan healthy setelah sekitar 20–30 detik).

Catatan: username dan password di atas hanya untuk belajar lokal. Jangan dipakai di server sungguhan.

Bab 4: Mengenal Management Dashboard#

Buka http://localhost:15672 dan login dengan admin / admin123.

Tab penting:

Di tab Queues, klik nama queue lalu buka bagian Get messages untuk mengintip isi pesan tanpa mengonsumsinya (pilih ack mode Nack message requeue true supaya pesan tidak hilang).

Arti kolom queue:

• Ready: pesan menunggu diambil consumer.
• Unacked: pesan sudah diambil consumer tapi belum di-ack.
• Total: Ready + Unacked.

Dashboard ini nanti dipakai untuk melihat langsung efek kode kita.

Bab 5: Struktur Project#

rabbitmq-react-lab/
├── docker-compose.yml
├── backend/
│   ├── package.json
│   └── src/
│       ├── config.js
│       ├── rabbit.js
│       ├── mongo.js
│       ├── api.js         ← producer + baca data
│       └── consumer.js    ← worker
└── frontend/              ← dibuat dengan Vite
    └── src/
        ├── App.jsx
        └── App.css

Setup backend:

mkdir -p backend/src
cd backend
npm init -y
npm install express cors amqplib mongodb

Edit backend/package.json: tambahkan "type": "module" (supaya bisa pakai import) dan script berikut:

{
  "name": "backend",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "api": "node --watch src/api.js",
    "consumer": "node --watch src/consumer.js"
  },
  "dependencies": {
    "amqplib": "^0.10.4",
    "cors": "^2.8.5",
    "express": "^4.21.0",
    "mongodb": "^6.9.0"
  }
}

Versi di atas hanya contoh. Biarkan npm install menulis versi terbaru yang kompatibel, kamu hanya perlu menambahkan "type" dan "scripts".

Bab 6: Backend: Helper Koneksi RabbitMQ#

6.1 src/config.js#

Semua konfigurasi di satu tempat, bisa dioverride lewat environment variable.

export const config = {
  rabbitUrl: process.env.RABBITMQ_URL || 'amqp://admin:admin123@localhost:5672',
  mongoUrl: process.env.MONGO_URL || 'mongodb://localhost:27017',
  mongoDb: process.env.MONGO_DB || 'rabbit_lab',
  collection: 'messages',
  queue: 'messages.queue',
  port: Number(process.env.PORT) || 3000,
  // Simulasi pekerjaan berat di consumer (ms), berguna untuk melihat queue menumpuk
  processDelayMs: Number(process.env.PROCESS_DELAY_MS) || 0,
};

6.2 src/rabbit.js#

Helper untuk connect dengan retry, karena RabbitMQ kadang belum siap saat aplikasi start.

import amqp from 'amqplib';
import { config } from './config.js';

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

/**
 * Connect ke RabbitMQ, ulangi sampai berhasil.
 */
export async function connectWithRetry(retries = 10, delayMs = 3000) {
  for (let attempt = 1; attempt <= retries; attempt++) {
    try {
      const connection = await amqp.connect(config.rabbitUrl);

      connection.on('error', (err) => {
        console.error('[rabbit] connection error:', err.message);
      });
      connection.on('close', () => {
        console.warn('[rabbit] connection closed');
      });

      console.log('[rabbit] connected');
      return connection;
    } catch (err) {
      console.error(`[rabbit] connect gagal (${attempt}/${retries}): ${err.message}`);
      if (attempt === retries) throw err;
      await sleep(delayMs);
    }
  }
}

/**
 * Deklarasi queue. Aman dipanggil berulang kali (idempotent),
 * asalkan parameternya SAMA persis di semua aplikasi.
 */
export async function assertMainQueue(channel) {
  await channel.assertQueue(config.queue, { durable: true });
}

Penting: jika assertQueue dipanggil dengan parameter berbeda dari queue yang sudah ada, RabbitMQ menolak dengan error PRECONDITION_FAILED. Karena itu, definisi queue dipusatkan di satu fungsi.

6.3 src/mongo.js#

import { MongoClient } from 'mongodb';
import { config } from './config.js';

let client;

export async function getCollection() {
  if (!client) {
    client = new MongoClient(config.mongoUrl);
    await client.connect();
    console.log('[mongo] connected');
  }
  return client.db(config.mongoDb).collection(config.collection);
}

Bab 7: Backend: API (Producer)#

File src/api.js punya tiga tugas:

1. POST /api/messages → validasi, lalu publish ke queue (bukan langsung ke MongoDB).
2. GET /api/messages → baca data dari MongoDB.
3. GET /api/queue-status → lihat jumlah pesan di queue (untuk demo).

import express from 'express';
import cors from 'cors';
import { randomUUID } from 'node:crypto';
import { config } from './config.js';
import { connectWithRetry, assertMainQueue } from './rabbit.js';
import { getCollection } from './mongo.js';

const app = express();
app.use(cors());
app.use(express.json());

// ---- Koneksi RabbitMQ (satu connection + satu confirm channel, dipakai ulang) ----
const connection = await connectWithRetry();
const channel = await connection.createConfirmChannel();
await assertMainQueue(channel);

/**
 * Publish dan tunggu konfirmasi dari broker (publisher confirm).
 * Promise resolve hanya jika RabbitMQ benar-benar menerima pesan.
 */
function publish(payload) {
  return new Promise((resolve, reject) => {
    const body = Buffer.from(JSON.stringify(payload));
    channel.sendToQueue(
      config.queue,
      body,
      {
        persistent: true,                 // simpan ke disk
        contentType: 'application/json',
        messageId: payload.id,
        timestamp: Date.now(),
      },
      (err) => (err ? reject(err) : resolve())
    );
  });
}

// ---- Routes ----

// 1. Kirim pesan ke queue
app.post('/api/messages', async (req, res) => {
  const { title, content } = req.body ?? {};

  if (typeof title !== 'string' || !title.trim()) {
    return res.status(400).json({ error: 'title wajib diisi' });
  }

  const message = {
    id: randomUUID(),
    title: title.trim(),
    content: typeof content === 'string' ? content.trim() : '',
    createdAt: new Date().toISOString(),
  };

  try {
    await publish(message);
    // 202 = diterima, belum tentu selesai diproses
    res.status(202).json({ status: 'queued', id: message.id });
  } catch (err) {
    console.error('[api] publish gagal:', err.message);
    res.status(503).json({ error: 'Gagal mengirim ke queue' });
  }
});

// 2. Ambil data yang SUDAH ditulis consumer ke MongoDB
app.get('/api/messages', async (_req, res) => {
  try {
    const col = await getCollection();
    const docs = await col.find().sort({ createdAt: -1 }).limit(50).toArray();
    res.json(docs.map(({ _id, ...rest }) => ({ id: _id, ...rest })));
  } catch (err) {
    console.error('[api] baca mongo gagal:', err.message);
    res.status(500).json({ error: 'Gagal membaca data' });
  }
});

// 3. Status queue (jumlah pesan menunggu dan jumlah consumer aktif)
app.get('/api/queue-status', async (_req, res) => {
  try {
    // checkQueue pada channel yang gagal akan menutup channel,
    // jadi pakai channel sementara.
    const ch = await connection.createChannel();
    const info = await ch.checkQueue(config.queue);
    await ch.close();
    res.json({
      queue: info.queue,
      waiting: info.messageCount,
      consumers: info.consumerCount,
    });
  } catch (err) {
    res.status(500).json({ error: err.message });
  }
});

app.listen(config.port, () => {
  console.log(`[api] listening on http://localhost:${config.port}`);
});

// Tutup koneksi dengan rapi saat proses dihentikan
process.on('SIGINT', async () => {
  await channel.close();
  await connection.close();
  process.exit(0);
});

Penjelasan penting#

• createConfirmChannel memungkinkan kita tahu bahwa RabbitMQ benar-benar menerima pesan (publisher confirm). Tanpa ini, sendToQueue hanya menaruh pesan di buffer lokal dan bisa hilang diam-diam kalau koneksi putus.
• 202 Accepted adalah status HTTP yang tepat: permintaan diterima tapi pemrosesan belum selesai.
• Satu koneksi dipakai ulang. Jangan membuat koneksi baru di setiap request. Koneksi TCP itu mahal.
• Pesan punya id unik (UUID). Kita pakai sebagai _id di MongoDB nanti supaya aman dari duplikasi (lihat Bab 11).

Bab 8: Backend: Consumer Worker#

File src/consumer.js mengambil pesan dari queue lalu menulisnya ke MongoDB.

import { config } from './config.js';
import { connectWithRetry, assertMainQueue } from './rabbit.js';
import { getCollection } from './mongo.js';

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

const connection = await connectWithRetry();
const channel = await connection.createChannel();
await assertMainQueue(channel);

// Maksimal 5 pesan "menggantung" (belum di-ack) pada consumer ini
await channel.prefetch(5);

const collection = await getCollection();

console.log(`[consumer] menunggu pesan di queue "${config.queue}" ...`);

await channel.consume(
  config.queue,
  async (msg) => {
    if (msg === null) return; // consumer dibatalkan oleh broker

    try {
      const data = JSON.parse(msg.content.toString());
      console.log(`[consumer] terima: ${data.id} - ${data.title}`);

      // Simulasi pekerjaan berat (opsional)
      if (config.processDelayMs > 0) await sleep(config.processDelayMs);

      // upsert dengan _id = id pesan → idempotent:
      // kalau pesan yang sama terkirim dua kali, data tidak dobel.
      await collection.updateOne(
        { _id: data.id },
        {
          $setOnInsert: {
            title: data.title,
            content: data.content,
            createdAt: data.createdAt,
            processedAt: new Date().toISOString(),
          },
        },
        { upsert: true }
      );

      channel.ack(msg); // beres → hapus dari queue
      console.log(`[consumer] selesai: ${data.id}`);
    } catch (err) {
      console.error('[consumer] gagal memproses:', err.message);

      // requeue=false → jangan diulang tanpa batas (infinite loop).
      // Di Bab 11 kita arahkan ke Dead Letter Queue.
      channel.nack(msg, false, false);
    }
  },
  { noAck: false } // WAJIB false supaya kita kontrol ack secara manual
);

process.on('SIGINT', async () => {
  await channel.close();
  await connection.close();
  process.exit(0);
});

Penjelasan penting#

• noAck: false berarti kita bertanggung jawab memanggil ack. Kalau proses crash di tengah jalan, pesan otomatis dikembalikan ke queue.
• ack dipanggil setelah MongoDB sukses. Urutan ini menjamin pesan tidak hilang. Konsekuensinya, pesan bisa diproses lebih dari sekali (misalnya crash tepat setelah menulis tapi sebelum ack). Itu sebabnya kita pakai upsert yang idempotent.
• nack(msg, false, false): argumen kedua allUpToHere=false, ketiga requeue=false. Tanpa Dead Letter Queue, pesan gagal akan dibuang. Kita perbaiki di Bab 11.
• PROCESS_DELAY_MS membuat consumer sengaja lambat supaya kamu bisa melihat queue menumpuk di dashboard.

Bab 9: Frontend React#

9.1 Buat project Vite#

Dari root rabbitmq-react-lab/:

npm create vite@latest frontend -- --template react
cd frontend
npm install

9.2 Proxy ke backend#

Agar tidak ribet urusan CORS saat development, edit frontend/vite.config.js:

import { defineConfig } from 'vite';
import react from '@vitejs/plugin-react';

export default defineConfig({
  plugins: [react()],
  server: {
    proxy: {
      '/api': 'http://localhost:3000',
    },
  },
});

Dengan ini, fetch('/api/messages') dari React otomatis diteruskan ke http://localhost:3000/api/messages.

9.3 src/App.jsx#

Fitur aplikasi:

• Form untuk mengirim satu pesan.
• Tombol Generate 10 data acak (sesuai “generate data from frontend” di Test Case 4).
• Panel status queue (jumlah pesan menunggu).
• Daftar data dari MongoDB, di-refresh otomatis tiap 2 detik.

import { useCallback, useEffect, useState } from 'react';
import './App.css';

const POLL_INTERVAL_MS = 2000;

const WORDS = ['alpha', 'bravo', 'charlie', 'delta', 'echo', 'foxtrot', 'golf', 'hotel'];
const pick = () => WORDS[Math.floor(Math.random() * WORDS.length)];

async function postMessage(title, content) {
  const res = await fetch('/api/messages', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ title, content }),
  });
  if (!res.ok) {
    const body = await res.json().catch(() => ({}));
    throw new Error(body.error || `HTTP ${res.status}`);
  }
  return res.json();
}

export default function App() {
  const [title, setTitle] = useState('');
  const [content, setContent] = useState('');
  const [messages, setMessages] = useState([]);
  const [queue, setQueue] = useState(null);
  const [sending, setSending] = useState(false);
  const [error, setError] = useState('');

  // Ambil data dari MongoDB + status queue
  const refresh = useCallback(async () => {
    try {
      const [msgRes, qRes] = await Promise.all([
        fetch('/api/messages'),
        fetch('/api/queue-status'),
      ]);
      if (msgRes.ok) setMessages(await msgRes.json());
      if (qRes.ok) setQueue(await qRes.json());
    } catch {
      // abaikan error polling sesaat (backend mungkin restart)
    }
  }, []);

  // Polling berkala
  useEffect(() => {
    refresh();
    const timer = setInterval(refresh, POLL_INTERVAL_MS);
    return () => clearInterval(timer); // cleanup, jangan lupa!
  }, [refresh]);

  const handleSubmit = async (e) => {
    e.preventDefault();
    setError('');
    setSending(true);
    try {
      await postMessage(title, content);
      setTitle('');
      setContent('');
    } catch (err) {
      setError(err.message);
    } finally {
      setSending(false);
    }
  };

  const handleGenerate = async () => {
    setError('');
    setSending(true);
    try {
      const jobs = Array.from({ length: 10 }, (_, i) =>
        postMessage(`Data ${pick()}-${Date.now() % 10000}-${i + 1}`, `Dibuat otomatis dari frontend`)
      );
      await Promise.all(jobs);
    } catch (err) {
      setError(err.message);
    } finally {
      setSending(false);
    }
  };

  return (
    <main className="container">
      <h1>🐰 RabbitMQ Lab</h1>
      <p className="subtitle">React → API → RabbitMQ → Consumer → MongoDB → React</p>

      <section className="card">
        <h2>Kirim pesan</h2>
        <form onSubmit={handleSubmit}>
          <input
            placeholder="Judul (wajib)"
            value={title}
            onChange={(e) => setTitle(e.target.value)}
            required
          />
          <textarea
            placeholder="Isi pesan (opsional)"
            value={content}
            onChange={(e) => setContent(e.target.value)}
            rows={3}
          />
          <div className="row">
            <button type="submit" disabled={sending}>
              {sending ? 'Mengirim...' : 'Kirim ke Queue'}
            </button>
            <button type="button" className="secondary" onClick={handleGenerate} disabled={sending}>
              Generate 10 data acak
            </button>
          </div>
        </form>
        {error && <p className="error">⚠ {error}</p>}
      </section>

      <section className="card">
        <h2>Status queue</h2>
        {queue ? (
          <div className="stats">
            <div>
              <strong>{queue.waiting}</strong>
              <span>menunggu</span>
            </div>
            <div>
              <strong>{queue.consumers}</strong>
              <span>consumer aktif</span>
            </div>
          </div>
        ) : (
          <p>Memuat...</p>
        )}
        {queue && queue.consumers === 0 && (
          <p className="warn">Belum ada consumer. Jalankan <code>npm run consumer</code>.</p>
        )}
      </section>

      <section className="card">
        <h2>Data di MongoDB ({messages.length})</h2>
        {messages.length === 0 ? (
          <p>Belum ada data.</p>
        ) : (
          <ul className="list">
            {messages.map((m) => (
              <li key={m.id}>
                <strong>{m.title}</strong>
                {m.content && <p>{m.content}</p>}
                <small>
                  dikirim {new Date(m.createdAt).toLocaleTimeString()} · diproses{' '}
                  {new Date(m.processedAt).toLocaleTimeString()}
                </small>
              </li>
            ))}
          </ul>
        )}
      </section>
    </main>
  );
}

9.4 src/App.css#

Hapus isi index.css bawaan Vite (atau biarkan), lalu ganti App.css:

body {
  margin: 0;
  font-family: system-ui, -apple-system, 'Segoe UI', sans-serif;
  background: #f4f5f7;
  color: #1f2937;
}

.container {
  max-width: 720px;
  margin: 0 auto;
  padding: 24px 16px 64px;
}

.subtitle {
  color: #6b7280;
  margin-top: -8px;
}

.card {
  background: #fff;
  border-radius: 12px;
  padding: 16px 20px;
  margin-bottom: 16px;
  box-shadow: 0 1px 3px rgba(0, 0, 0, 0.08);
}

.card h2 {
  margin-top: 0;
  font-size: 1.1rem;
}

form {
  display: flex;
  flex-direction: column;
  gap: 10px;
}

input,
textarea {
  padding: 10px;
  border: 1px solid #d1d5db;
  border-radius: 8px;
  font: inherit;
}

.row {
  display: flex;
  gap: 8px;
  flex-wrap: wrap;
}

button {
  padding: 10px 16px;
  border: 0;
  border-radius: 8px;
  background: #ff6600;
  color: #fff;
  font: inherit;
  cursor: pointer;
}

button.secondary {
  background: #374151;
}

button:disabled {
  opacity: 0.6;
  cursor: not-allowed;
}

.stats {
  display: flex;
  gap: 32px;
}

.stats div {
  display: flex;
  flex-direction: column;
}

.stats strong {
  font-size: 2rem;
  color: #ff6600;
}

.list {
  list-style: none;
  padding: 0;
  margin: 0;
}

.list li {
  padding: 10px 0;
  border-bottom: 1px solid #e5e7eb;
}

.list li:last-child {
  border-bottom: 0;
}

.list p {
  margin: 4px 0;
}

.list small {
  color: #6b7280;
}

.error {
  color: #b91c1c;
}

.warn {
  color: #b45309;
}

Bab 10: Menjalankan dan Menguji#

Buka tiga terminal (infrastruktur Docker sudah jalan dari Bab 3).

Terminal 1: API

cd backend
npm run api

Terminal 2: Frontend

cd frontend
npm run dev

Buka URL yang tampil (biasanya http://localhost:5173).

Terminal 3: Consumer. Jangan dijalankan dulu, kita sengaja melihat perilaku queue.

Eksperimen 1: Pesan menunggu di queue#

1. Di frontend, klik Generate 10 data acak.
2. Lihat panel Status queue: menunggu: 10, consumer aktif: 0. Daftar MongoDB masih kosong.
3. Buka dashboard RabbitMQ → tab Queues → messages.queue. Kolom Ready = 10.

Ini membuktikan pesan aman tersimpan walau consumer belum ada.

Eksperimen 2: Consumer mulai bekerja#

Jalankan consumer di terminal 3:

cd backend
npm run consumer

Dalam hitungan detik: log consumer menampilkan 10 pesan diproses, panel menunggu turun ke 0, dan daftar di frontend terisi data dari MongoDB. 🎉

Eksperimen 3: Beban lambat#

Hentikan consumer (Ctrl+C), lalu jalankan dengan delay 2 detik per pesan:

# Linux / macOS
PROCESS_DELAY_MS=2000 npm run consumer

# Windows PowerShell
$env:PROCESS_DELAY_MS=2000; npm run consumer

Klik Generate 10 data acak lagi. Perhatikan di dashboard: Unacked naik sampai 5 (sesuai prefetch(5)), sisanya di Ready. Angka itu turun bertahap seiring pesan selesai.

Eksperimen 4: Consumer crash#

Saat consumer lambat tadi memproses, matikan paksa dengan Ctrl+C. Pesan Unacked otomatis kembali ke Ready. Jalankan consumer lagi dan semua pesan tetap terproses. Tidak ada yang hilang.

Eksperimen 5: Scale out#

Jalankan dua consumer sekaligus di dua terminal. Klik Generate beberapa kali. RabbitMQ membagi pesan ke kedua consumer secara bergantian (round-robin). Ini cara menaikkan throughput.

Cek data di MongoDB#

Pakai MongoDB Compass dengan koneksi mongodb://localhost:27017, database rabbit_lab, collection messages. Atau lewat terminal:

docker exec -it lab-mongo mongosh rabbit_lab --eval "db.messages.find().limit(3)"

Bab 11: Konsep Lanjutan: Ack, Retry, dan Dead Letter Queue#

Di Bab 8, pesan yang gagal di-nack tanpa requeue langsung hilang. Di produksi, kita ingin pesan gagal disimpan untuk diperiksa. Solusinya Dead Letter Queue (DLQ).

11.1 Konsep#

messages.queue ──(nack/ditolak/expired)──► dlx exchange ──► messages.dlq

Pesan “mati” (di-nack tanpa requeue, kedaluwarsa, atau queue penuh) dialihkan ke Dead Letter Exchange (DLX), lalu masuk ke queue khusus yang bisa kamu periksa dan proses ulang.

11.2 Implementasi#

Ubah assertMainQueue di src/rabbit.js:

export const DLX = 'messages.dlx';
export const DLQ = 'messages.dlq';

export async function assertMainQueue(channel) {
  // 1. Exchange + queue untuk pesan gagal
  await channel.assertExchange(DLX, 'direct', { durable: true });
  await channel.assertQueue(DLQ, { durable: true });
  await channel.bindQueue(DLQ, DLX, config.queue);

  // 2. Queue utama, diarahkan ke DLX bila ada pesan yang ditolak
  await channel.assertQueue(config.queue, {
    durable: true,
    arguments: {
      'x-dead-letter-exchange': DLX,
      'x-dead-letter-routing-key': config.queue,
    },
  });
}

⚠️ Queue yang sudah ada tidak bisa diubah argumennya. Kamu akan mendapat error PRECONDITION_FAILED. Hapus dulu queue lama lewat dashboard (Queues → messages.queue → Delete), baru jalankan ulang API dan consumer.

11.3 Uji DLQ#

Tambahkan sementara pemicu error di consumer, di dalam try sebelum penulisan ke MongoDB:

if (data.title.toLowerCase().includes('error')) {
  throw new Error('Simulasi gagal');
}

Kirim pesan dengan judul error test. Pesan akan muncul di messages.dlq pada dashboard. Hapus baris simulasi setelah selesai mencoba.

11.4 Retry dengan batas percobaan#

Pola sederhana: hitung berapa kali pesan sudah dicoba lewat header. Pesan yang di-nack dengan requeue=false dan masuk DLX otomatis mendapat header x-death berisi hitungan. Untuk belajar, pola manual berikut cukup dan mudah dipahami:

const MAX_RETRY = 3;

// di dalam catch consumer:
const retry = (msg.properties.headers?.['x-retry'] ?? 0) + 1;

if (retry <= MAX_RETRY) {
  // kirim ulang ke queue yang sama dengan counter bertambah, lalu ack yang lama
  channel.sendToQueue(config.queue, msg.content, {
    persistent: true,
    contentType: 'application/json',
    messageId: msg.properties.messageId,
    headers: { 'x-retry': retry },
  });
  channel.ack(msg);
} else {
  channel.nack(msg, false, false); // menyerah → masuk DLQ
}

Untuk retry dengan jeda (misalnya 10 detik, lalu 1 menit), pola produksi memakai retry queue dengan x-message-ttl yang di-dead-letter kembali ke queue utama. Ini topik latihan di Bab 16.

11.5 Idempotency: wajib untuk consumer#

RabbitMQ menjamin at-least-once delivery: pesan bisa sampai lebih dari sekali (consumer crash sebelum ack, retry, dsb.). Jadi consumer harus idempotent: memproses pesan yang sama dua kali tidak boleh menghasilkan efek ganda.

Di ebook ini kita sudah menerapkannya: _id = id pesan + upsert + $setOnInsert. Kirim pesan yang sama dua kali, dokumen di MongoDB tetap satu.

Bab 12: Exchange: Direct, Fanout, Topic#

Sejauh ini kita memakai default exchange. Untuk skenario yang lebih kompleks, pakai exchange eksplisit.

12.1 Jenis exchange#

12.2 Contoh topic exchange#

Misal kita ingin satu event masuk ke beberapa consumer berbeda:

// Setup (producer dan consumer sama-sama memanggil ini)
await ch.assertExchange('events', 'topic', { durable: true });

// Consumer A: hanya tertarik pada order.created
await ch.assertQueue('q.email', { durable: true });
await ch.bindQueue('q.email', 'events', 'order.created');

// Consumer B: semua event order
await ch.assertQueue('q.audit', { durable: true });
await ch.bindQueue('q.audit', 'events', 'order.#');

// Producer
ch.publish(
  'events',
  'order.created',
  Buffer.from(JSON.stringify({ orderId: 1 })),
  { persistent: true }
);

Hasilnya:

• order.created → masuk q.email dan q.audit.
• order.cancelled → hanya masuk q.audit.

12.3 Kapan memakai apa?#

• Satu tugas, satu antrean, banyak worker: seperti ebook ini (work queue, default exchange).
• Satu event, banyak penerima berbeda: fanout atau topic (publish/subscribe).

Bab 13: Best Practice#

Pesan dan queue

• Gunakan durable: true + persistent: true untuk data penting.
• Selalu pakai manual ack (noAck: false).
• Selalu atur prefetch. Nilai 1–10 cocok untuk pekerjaan berat, lebih besar untuk pekerjaan ringan.
• Pesan kecil dan berisi ID lebih baik daripada payload besar. Untuk file besar, simpan di object storage (misalnya OBS) dan kirim referensinya.
• Beri setiap pesan ID unik untuk idempotency dan tracing.
• Pakai DLQ untuk semua queue produksi.

Koneksi

• Satu koneksi per proses, banyak channel kalau perlu. Jangan buka koneksi per request.
• Aktifkan publisher confirms pada producer.
• Tangani event error dan close pada koneksi, dan siapkan logika reconnect.

Keamanan

• Jangan pakai user guest/admin123 di luar lokal.
• Simpan kredensial di environment variable atau secret manager, bukan di kode.
• Batasi akses dashboard (port 15672) hanya dari jaringan internal atau VPN.

Frontend

• Frontend tidak boleh terhubung langsung ke RabbitMQ. Selalu lewat API. Kredensial broker tidak boleh sampai ke browser.
• Polling cukup untuk belajar. Untuk produksi, pertimbangkan Server-Sent Events atau WebSocket agar update real-time tanpa polling terus-menerus.

Bab 14: Troubleshooting#

Perintah berguna:

docker logs -f lab-rabbitmq                       # log RabbitMQ
docker exec lab-rabbitmq rabbitmqctl list_queues   # daftar queue via CLI
docker compose down -v                             # reset total (hapus data!)

Bab 15: Migrasi ke FunctionGraph#

Ketika FunctionGraph sudah tersedia, arsitektur tetap sama. Yang berubah adalah cara kodenya dijalankan.

15.1 Pemetaan komponen#

Nama layanan dan jenis trigger yang persis tersedia dapat berbeda antar region dan versi. Cek dokumentasi resmi Huawei Cloud untuk detailnya.

15.2 Perbedaan cara berpikir#

Producer function: sifatnya stateless, jadi koneksi sebaiknya dibuat di luar handler agar bisa dipakai ulang antar invokasi (warm start):

// dibuat sekali per instance function
let channelPromise;
const getChannel = () => (channelPromise ??= init());

export const handler = async (event, context) => {
  const ch = await getChannel();
  // ... validasi event.body lalu publish seperti di api.js
  return { statusCode: 202, body: JSON.stringify({ status: 'queued' }) };
};

Consumer function: tidak ada lagi channel.consume. Platform memanggil function-mu dengan batch pesan dalam event. Kamu cukup memproses dan, bila sukses, platform yang menangani konfirmasi. Logika intinya, yaitu upsert idempotent ke MongoDB, sama persis dengan consumer.js.

export const handler = async (event) => {
  const col = await getCollection();
  for (const record of event.records ?? []) {
    const data = JSON.parse(record.body);
    await col.updateOne({ _id: data.id }, { $setOnInsert: { ...data, processedAt: new Date().toISOString() } }, { upsert: true });
  }
  return { ok: true };
};

Bentuk event persis (nama field seperti records/body) tergantung konfigurasi trigger. Lihat contoh event di console FunctionGraph saat membuat trigger.

15.3 Hal yang perlu diperhatikan#

• Timeout function terbatas. Pekerjaan berat harus dipecah.
• Concurrency: banyak instance function bisa jalan bersamaan. Idempotency (Bab 11.5) jadi semakin penting.
• Koneksi MongoDB: reuse koneksi di luar handler. Jangan connect ulang tiap invokasi.
• Networking: function harus berada di VPC yang bisa menjangkau RabbitMQ dan MongoDB.
• Frontend: ubah base URL API dari proxy lokal ke endpoint APIG, lewat environment variable build (VITE_API_BASE).

Bab 16: Latihan Mandiri#

Urut dari mudah ke sulit.

1. Tombol hapus. Tambahkan DELETE /api/messages/:id dan tombol di React.
2. Ubah prefetch. Coba prefetch(1) vs prefetch(50) dengan dua consumer lambat. Amati pembagian beban di dashboard.
3. Status per pesan. Setelah POST, frontend menampilkan item dengan status “⏳ dalam antrean”. Setelah muncul di MongoDB, berubah menjadi “✅ selesai”. (Petunjuk: simpan id hasil POST di state, lalu cocokkan dengan data MongoDB.)
4. Server-Sent Events. Ganti polling dengan SSE: API mengirim event saat consumer selesai menulis. (Petunjuk: consumer mem-publish event ke fanout exchange, API subscribe lalu meneruskan ke browser lewat SSE.)
5. Retry dengan jeda. Buat messages.retry dengan x-message-ttl: 10000 yang di-dead-letter kembali ke messages.queue, sehingga pesan gagal dicoba ulang setelah 10 detik, maksimal 3 kali.
6. Reprocess DLQ. Buat endpoint admin yang memindahkan pesan dari DLQ kembali ke queue utama.
7. Topic exchange. Tambahkan jenis pesan (type: 'info' | 'warning') dan dua consumer terpisah: satu menyimpan semuanya, satu hanya warning.
8. Login Keycloak. Tambahkan keycloak-js di React (Test Case 1 pada diagram), kirim access token di header Authorization, dan verifikasi JWT di API sebelum publish.
9. Test beban. Kirim 10.000 pesan lewat script. Ukur waktu sampai semuanya masuk MongoDB dengan 1, 2, dan 4 consumer.
10. Monitoring. Aktifkan plugin Prometheus pada RabbitMQ (rabbitmq_prometheus) dan buat dashboard sederhana.

Bab 17: Penutup dan Cheat Sheet#

17.1 Yang sudah kamu pelajari#

• Konsep producer, exchange, queue, binding, consumer.
• Membangun pipeline lengkap: React → API → RabbitMQ → Consumer → MongoDB → React.
• Ack, prefetch, durability, persistence, publisher confirm.
• DLQ, retry, dan idempotency.
• Jenis exchange dan kapan memakainya.
• Cara memetakan semuanya ke FunctionGraph.

17.2 Cheat sheet amqplib#

// Koneksi
const conn = await amqp.connect(url);
const ch = await conn.createChannel();          // channel biasa
const cch = await conn.createConfirmChannel();  // channel dengan publisher confirm

// Deklarasi
await ch.assertQueue('q', { durable: true });
await ch.assertExchange('ex', 'topic', { durable: true });
await ch.bindQueue('q', 'ex', 'pattern.#');

// Kirim
ch.sendToQueue('q', Buffer.from('...'), { persistent: true });
ch.publish('ex', 'routing.key', Buffer.from('...'), { persistent: true });

// Terima
await ch.prefetch(5);
await ch.consume('q', (msg) => { /* proses */ ch.ack(msg); }, { noAck: false });

// Respons
ch.ack(msg);
ch.nack(msg, false, false);   // allUpTo=false, requeue=false
ch.reject(msg, false);        // sama, untuk satu pesan

// Info & penutup
await ch.checkQueue('q');     // { messageCount, consumerCount }
await ch.purgeQueue('q');     // kosongkan queue
await ch.close();
await conn.close();

17.3 Aturan emas#

1. Producer tidak menunggu pekerjaan selesai, hanya menunggu confirm bahwa pesan tersimpan.
2. Consumer harus idempotent.
3. ack setelah pekerjaan selesai, bukan sebelum.
4. Pesan gagal harus punya tempat tujuan (DLQ), jangan dibuang diam-diam.
5. Frontend tidak pernah bicara langsung ke broker.

17.4 Sumber belajar lanjutan#

• Dokumentasi resmi RabbitMQ: https://www.rabbitmq.com/docs
• Tutorial resmi RabbitMQ (JavaScript): https://www.rabbitmq.com/tutorials
• Dokumentasi amqplib: https://amqp-node.github.io/amqplib/
• Dokumentasi MongoDB Node.js Driver: https://www.mongodb.com/docs/drivers/node/current/

Selamat belajar! Kalau sudah nyaman dengan versi lokal ini, pindah ke FunctionGraph hanya soal membungkus logika yang sama dalam handler. 🐇