Event log dan pub/sub pada chapter sebelumnya hidup di dalam satu proses — begitu prosesnya berhenti, seluruh log ikut hilang; begitu dua service berjalan di dua mesin, tidak ada lagi satu objek Bus bersama. Message broker memindahkan mesin itu keluar dari aplikasi: satu proses terpisah yang menerima pesan dari penerbit, menyimpannya dengan awet, lalu menyampaikannya kepada pelanggan — dengan jaminan urutan, pengiriman ulang, dan skala yang tidak mungkin dicapai in-process. Kafka dan RabbitMQ adalah dua nama yang paling sering dijumpai, dari dua filosofi berbeda. Chapter ini membangun mekanisme intinya — partisi, urutan per kunci, dead letter queue — dalam Java murni, lalu memetakan lanskap broker produksi.

Apa yang Dikerjakan Broker #

Tanpa broker, setiap pasang service harus membangun sendiri daftar kerja berikut — salah satunya gagal berarti pesan hilang:

Tanggung jawabTanpa brokerDengan broker
Menyimpan pesan saat pelanggan sedang matitidak tersimpanpesan menunggu di broker
Menjamin urutankoordinasi manualFIFO per antrean/partisi
Menyebarkan ke banyak pelanggandaftar alamat manuallangganan per topik/exchange
Pengiriman ulang setelah kegagalantidak adaat-least-once via acknowledgment
Menampung lonjakan bebanpemanggil menanggungantrean meredam lonjakan

Perhatikan baris terakhir: broker adalah penampung guncangan — ketika svc-pembayaran mendadak lambat, antrean memanjang, dan kedalamannya menjadi sinyal operasional bahwa service perlu diperbaiki atau discale.

Partisi Kunci dan Urutan Pesan #

Menjamin urutan global membutuhkan satu antrean tunggal — bottleneck yang membatalkan manfaat terdistribusi. Solusi Kafka: partisi berdasarkan kunci. Pesan berkunci sama selalu jatuh ke partisi yang sama, sehingga urutan per kunci terjaga, sementara partisi berbeda berjalan paralel. Program berikut membangun mekanisme itu ditambah dead letter queue untuk pesan yang terus gagal:

java AntreanPesan.java 
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;

/** Broker mini: partisi per kunci (urutan per kunci terjaga) + dead letter queue. */
public class AntreanPesan {

    /** Satu pesan: kunci penentu partisi + isi. */
    record Pesan(String kunci, String isi) {}

    /** Broker dengan jumlah partisi tetap dan dead letter queue (DLQ). */
    static class Broker {
        private final List<Deque<Pesan>> partisi;
        private final Deque<Pesan> deadLetter = new ArrayDeque<>();

        Broker(int jumlahPartisi) {
            partisi = new ArrayList<>();
            for (int i = 0; i < jumlahPartisi; i++) {
                partisi.add(new ArrayDeque<>());
            }
        }

        /** Kunci yang sama selalu jatuh ke partisi yang sama. */
        int kirim(Pesan p) {
            int idx = Math.floorMod(p.kunci().hashCode(), partisi.size());
            partisi.get(idx).offer(p); // FIFO: offer di ekor
            return idx;
        }

        /** Konsumsi satu partisi; pesan bermasalah di-retry lalu dibuang ke DLQ. */
        void konsumsi(int idx) {
            Deque<Pesan> antrean = partisi.get(idx);
            while (!antrean.isEmpty()) {
                Pesan p = antrean.poll(); // FIFO: poll dari kepala
                int percobaan = 0;
                boolean sukses = false;
                while (percobaan < 3 && !sukses) {
                    percobaan++;
                    sukses = !p.isi().contains("RACUN");
                }
                if (sukses) {
                    System.out.println("  partisi " + idx + ": [" + p.kunci() + "] "
                            + p.isi() + " -> diproses (percobaan ke-" + percobaan + ")");
                } else {
                    deadLetter.add(p);
                    System.out.println("  partisi " + idx + ": [" + p.kunci() + "] "
                            + p.isi() + " -> gagal " + percobaan
                            + " percobaan, dipindah ke DLQ");
                }
            }
        }

        Deque<Pesan> deadLetter() {
            return deadLetter;
        }
    }

    public static void main(String[] args) {
        Broker broker = new Broker(2); // dua partisi, seperti topik Kafka

        System.out.println("== Penempatan partisi: kunci sama -> partisi sama ==");
        List<Pesan> pesanan = List.of(
                new Pesan("pelanggan-7", "perbarui alamat"),
                new Pesan("pelanggan-7", "aktifkan langganan"),
                new Pesan("pelanggan-7", "kirim faktur"),
                new Pesan("pelanggan-2", "perbarui alamat"),
                new Pesan("pelanggan-7", "RACUN: isi tidak bisa diproses"),
                new Pesan("pelanggan-2", "naikkan paket"));

        for (Pesan p : pesanan) {
            int idx = broker.kirim(p);
            System.out.println("  [" + p.kunci() + "] " + p.isi() + " -> partisi " + idx);
        }

        System.out.println();
        System.out.println("== Konsumsi per partisi (urutan FIFO terjaga) ==");
        broker.konsumsi(0);
        broker.konsumsi(1);

        System.out.println();
        System.out.println("== Dead letter queue: pesan racun terisolasi ==");
        for (Pesan p : broker.deadLetter()) {
            System.out.println("  [" + p.kunci() + "] " + p.isi());
        }
    }
}

Output yang Dihasilkan:

text 
== Penempatan partisi: kunci sama -> partisi sama ==
  [pelanggan-7] perbarui alamat -> partisi 1
  [pelanggan-7] aktifkan langganan -> partisi 1
  [pelanggan-7] kirim faktur -> partisi 1
  [pelanggan-2] perbarui alamat -> partisi 0
  [pelanggan-7] RACUN: isi tidak bisa diproses -> partisi 1
  [pelanggan-2] naikkan paket -> partisi 0

== Konsumsi per partisi (urutan FIFO terjaga) ==
  partisi 0: [pelanggan-2] perbarui alamat -> diproses (percobaan ke-1)
  partisi 0: [pelanggan-2] naikkan paket -> diproses (percobaan ke-1)
  partisi 1: [pelanggan-7] perbarui alamat -> diproses (percobaan ke-1)
  partisi 1: [pelanggan-7] aktifkan langganan -> diproses (percobaan ke-1)
  partisi 1: [pelanggan-7] kirim faktur -> diproses (percobaan ke-1)
  partisi 1: [pelanggan-7] RACUN: isi tidak bisa diproses -> gagal 3 percobaan, dipindah ke DLQ

== Dead letter queue: pesan racun terisolasi ==
  [pelanggan-7] RACUN: isi tidak bisa diproses

Mari kita bedah tiga mekanismenya:

  1. kirim(...) dengan Math.floorMod(p.kunci().hashCode(), partisi.size()): hash kunci menentukan partisi secara stabil — pelanggan-7 selalu jatuh ke partisi 1, pelanggan-2 selalu ke partisi 0, pada setiap pengiriman. floorMod menjaga hasil tetap positif meski hash negatif — kontrak inilah yang menjaga urutan per pelanggan: seluruh pesannya mengantre di satu jalur.
  2. Deque.offer dan poll: ArrayDeque sebagai antrean FIFO — offer menambah di ekor, poll mengambil dari kepala. Pada keluaran, keempat pesan pelanggan-7 terkonsumsi persis dalam urutan pengiriman, termasuk pesan racun yang tetap berada di posisi kelimanya.
  3. Loop tiga percobaan dan deadLetter: pesan berisi RACUN selalu gagal diproses. Setelah tiga percobaan, pesan dipindahkan ke DLQ — antrean terpisah untuk pesan bermasalah — sehingga pesan sehat di belakangnya tidak tersumbat selamanya. Membaca DLQ dan memperbaiki penyebabnya adalah rutinitas operasional tiap sistem messaging.

Kafka dan RabbitMQ Dua Keluarga #

Kedua broker menyelesaikan daftar kerja di atas dengan bentuk yang berbeda:

AspekApache KafkaRabbitMQ
Model intilog ter-append per partisiexchange + routing ke antrean
Konsumsipelanggan menarik (pull) dengan offsetbroker mendorong (push) ke konsumen
Pesan setelah dikonsumsitetap di log sampai retensi habis — bisa di-replaydihapus setelah di-acknowledge
Retensi bawaanberhari-hari hingga berminggu-minggutidak ada (antrean kosong = hilang)
Pola khasevent streaming, event sourcing, pipelinetask queue, RPC asinkron, routing rumit
Urutanper partisi (per kunci)per antrean

Perbedaan paling mendasar ada di baris “pesan setelah dikonsumsi”: Kafka menyimpan sejarah — pelanggan baru bisa memutar ulang masa lalu seperti fungsi putarUlang pada chapter sebelumnya — sedangkan RabbitMQ memperlakukan pesan sebagai kiriman sekali jalan. Karena itu Kafka wajar untuk arsitektur event-driven yang kita bangun (event = fakta yang bisa dibaca ulang), dan RabbitMQ kuat untuk penyebaran tugas dengan routing halus antar banyak antrean.

Klien Java Menyambung ke Broker #

Di Java, Kafka diakses lewat pustaka resmi kafka-clients, dan RabbitMQ lewat amqp-client — keduanya adalah dependensi Maven/Gradle, bukan bagian JDK, sehingga contoh di sini tampil sebagai konfigurasi tanpa klaim eksekusi:

java Konfigurasi produsen dan konsumen Kafka — potongan konfigurasi 
Properties props = new Properties();
props.put("bootstrap.servers", "broker1.internal:9092,broker2.internal:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> produsen = new KafkaProducer<>(props);
produsen.send(new ProducerRecord<>("pesanan-events", "pelanggan-7", "kirim faktur"));
java Konsumen dengan consumer group — potongan konfigurasi 
props.put("group.id", "svc-notifikasi");
props.put("enable.auto.commit", "false");           // offset dikomit setelah diproses
props.put("auto.offset.reset", "earliest");        // anggota baru mulai dari awal log

KafkaConsumer<String, String> konsumen = new KafkaConsumer<>(props);
konsumen.subscribe(List.of("pesanan-events"));
// poll() mengembalikan record; commitSync() menandai offset setelah diproses

Dua detail itu menghubungkan chapter sebelumnya: group.id membuat sekelompok konsumen membagi partisi (skala paralel), dan commit offset setelah diproses itulah yang menghasilkan semantik at-least-once — konsumen yang jatuh sebelum komit menerima ulang rentang pesan, sehingga handler wajib idempotent.

Broker menjawab “bagaimana pesan berpindah antar service”. Pertanyaan konfigurasi yang tersisa pada arsitektur terdistribusi: bagaimana sebuah service menemukan alamat service lain yang jumlah replikanya berubah-ubah? Chapter berikutnya membahas service discovery dan registry dengan detak jantung.

Referensi #

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

AntreanPesan.java

Unduh berkas
java 
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;

/** Broker mini: partisi per kunci (urutan per kunci terjaga) + dead letter queue. */
public class AntreanPesan {

    /** Satu pesan: kunci penentu partisi + isi. */
    record Pesan(String kunci, String isi) {}

    /** Broker dengan jumlah partisi tetap dan dead letter queue (DLQ). */
    static class Broker {
        private final List<Deque<Pesan>> partisi;
        private final Deque<Pesan> deadLetter = new ArrayDeque<>();

        Broker(int jumlahPartisi) {
            partisi = new ArrayList<>();
            for (int i = 0; i < jumlahPartisi; i++) {
                partisi.add(new ArrayDeque<>());
            }
        }

        /** Kunci yang sama selalu jatuh ke partisi yang sama. */
        int kirim(Pesan p) {
            int idx = Math.floorMod(p.kunci().hashCode(), partisi.size());
            partisi.get(idx).offer(p); // FIFO: offer di ekor
            return idx;
        }

        /** Konsumsi satu partisi; pesan bermasalah di-retry lalu dibuang ke DLQ. */
        void konsumsi(int idx) {
            Deque<Pesan> antrean = partisi.get(idx);
            while (!antrean.isEmpty()) {
                Pesan p = antrean.poll(); // FIFO: poll dari kepala
                int percobaan = 0;
                boolean sukses = false;
                while (percobaan < 3 && !sukses) {
                    percobaan++;
                    sukses = !p.isi().contains("RACUN");
                }
                if (sukses) {
                    System.out.println("  partisi " + idx + ": [" + p.kunci() + "] "
                            + p.isi() + " -> diproses (percobaan ke-" + percobaan + ")");
                } else {
                    deadLetter.add(p);
                    System.out.println("  partisi " + idx + ": [" + p.kunci() + "] "
                            + p.isi() + " -> gagal " + percobaan
                            + " percobaan, dipindah ke DLQ");
                }
            }
        }

        Deque<Pesan> deadLetter() {
            return deadLetter;
        }
    }

    public static void main(String[] args) {
        Broker broker = new Broker(2); // dua partisi, seperti topik Kafka

        System.out.println("== Penempatan partisi: kunci sama -> partisi sama ==");
        List<Pesan> pesanan = List.of(
                new Pesan("pelanggan-7", "perbarui alamat"),
                new Pesan("pelanggan-7", "aktifkan langganan"),
                new Pesan("pelanggan-7", "kirim faktur"),
                new Pesan("pelanggan-2", "perbarui alamat"),
                new Pesan("pelanggan-7", "RACUN: isi tidak bisa diproses"),
                new Pesan("pelanggan-2", "naikkan paket"));

        for (Pesan p : pesanan) {
            int idx = broker.kirim(p);
            System.out.println("  [" + p.kunci() + "] " + p.isi() + " -> partisi " + idx);
        }

        System.out.println();
        System.out.println("== Konsumsi per partisi (urutan FIFO terjaga) ==");
        broker.konsumsi(0);
        broker.konsumsi(1);

        System.out.println();
        System.out.println("== Dead letter queue: pesan racun terisolasi ==");
        for (Pesan p : broker.deadLetter()) {
            System.out.println("  [" + p.kunci() + "] " + p.isi());
        }
    }
}