Message Queues

Flash sale 17 Agustus menguji batas TokoKita: pada puncaknya aplikasi melepas ratusan event pesanan per detik, sementara pipeline analitik hanya sanggup mencerna puluhan. Tanpa penyangga, aplikasi harus menunggu pipeline atau, lebih buruk, event gagal tersimpan sama sekali. Message queue adalah komponen tengah yang menjawab masalah ini: producer menaruh pesan, consumer mengambil sesuai kecepatannya sendiri, dan keduanya tidak perlu saling mengenal. Chapter ini membedah peran antrean pesan, dua pola distribusinya, dan fenomena backpressure yang kita buktikan dengan eksperimen thread sungguhan.

Peran Antrian Pesan dalam Sistem Data #

Antrean pesan memberi tiga jaminan praktis. Pertama, decoupling: layanan pesanan cukup tahu alamat antrean, bukan daftar sepuluh sistem pemakai datanya. Kedua, buffering: ketika laju produksi melebihi laju konsumsi, antrean menampung sementara selisihnya sehingga lonjakan terserap tanpa menolak pesan. Ketiga, asinkron: producer selesai bekerja begitu pesan masuk antrean, tanpa menunggu consumer menyelesaikan pemrosesan.

Jaminan tersebut berlaku hanya jika pesan bertahan saat salah satu pihak berhenti. Antrean dalam memori pada satu proses — seperti simulasi event bus pada chapter sebelumnya — ikut hilang bersama prosesnya. Queue produksi karena itu berjalan sebagai broker terpisah yang menyimpan pesan di disk dan menyediakan mekanisme posisi baca, sehingga consumer yang direstart melanjutkan dari pesan terakhir yang belum diproses.

Aspek	Point-to-point	Publish-subscribe
Penerima satu pesan	tepat satu consumer	semua subscriber aktif
Tujuan khas	pembagian tugas (task queue)	penyebaran event
Urutan kerja	satu pesan diproses sekali	tiap subscriber memproses salinannya
Contoh TokoKita	antrean kirim email transaksional	event `order.dibuat` ke stok dan analitik

Dua pola tersebut tidak saling meniadakan. Kafka menyatukan keduanya lewat consumer group: setiap pesan dalam satu partition hanya diterima satu anggota group (perilaku point-to-point), sementara group berbeda masing-masing menerima salinan lengkap (perilaku publish-subscribe). Mekanisme pembagiannya bergantung pada konsep partition dan offset — bahasan dua chapter berikutnya.

Backpressure saat Consumer Lebih Lambat #

Backpressure adalah sinyal “perlambat” yang mengalir dari consumer kembali ke producer. Bentuk paling mudah diamati ada pada bounded queue: antrean dengan kapasitas maksimum, di mana put() akan memblokir begitu antrean penuh. Eksperimen berikut menjalankan producer cepat dan consumer lambat dalam dua skenario — antrean berbatas (maxsize=3) dan tanpa batas — lalu membandingkan waktu tunggu, kedalaman antrean, dan kesegaran pesan terakhir.

python
"""Backpressure: bounded queue menahan producer saat consumer lebih lambat.
Dua skenario dibandingkan: maxsize=3 (bounded) vs maxsize=0 (unbounded)."""

import threading
import time
from queue import Queue

JUMLAH_PESAN = 10
LAMBAT_CONSUMER = 0.05  # detik per pesan


def jalankan_skenario(maxsize, nama):
    q = Queue(maxsize=maxsize)
    kirim_pada = {}
    stat = {"tunggu_producer": 0.0, "tunggu_pesan_terakhir": 0.0, "kedalaman_maks": 0}

    def producer():
        for nomor in range(1, JUMLAH_PESAN + 1):
            pesan = f"P-{nomor:02d}"
            mulai = time.perf_counter()
            q.put(pesan)  # memblok bila antrean penuh (maxsize tercapai)
            stat["tunggu_producer"] += time.perf_counter() - mulai
            kirim_pada[pesan] = time.perf_counter()
            stat["kedalaman_maks"] = max(stat["kedalaman_maks"], q.qsize())

    def consumer():
        while True:
            pesan = q.get()
            if pesan is None:
                return
            time.sleep(LAMBAT_CONSUMER)  # pemrosesan sengaja lebih lambat
            stat["tunggu_pesan_terakhir"] = time.perf_counter() - kirim_pada[pesan]

    mulai = time.perf_counter()
    t_consumer = threading.Thread(target=consumer)
    t_consumer.start()
    t_producer = threading.Thread(target=producer)
    t_producer.start()
    t_producer.join()
    q.put(None)  # sinyal berhenti untuk consumer
    t_consumer.join()
    durasi = time.perf_counter() - mulai

    print(nama)
    print(f"  producer total menunggu  : {stat['tunggu_producer'] * 1000:6.1f} ms")
    print(f"  kedalaman antrean maks   : {stat['kedalaman_maks']} pesan")
    print(f"  pesan terakhir menunggu  : {stat['tunggu_pesan_terakhir'] * 1000:6.1f} ms "
          f"sebelum diproses")
    print(f"  durasi total             : {durasi * 1000:6.1f} ms\n")


def main():
    jalankan_skenario(maxsize=3, nama="Skenario A: bounded queue (maxsize=3)")
    jalankan_skenario(maxsize=0, nama="Skenario B: unbounded queue (maxsize=0)")


if __name__ == "__main__":
    main()

Output yang Dihasilkan:

text
Skenario A: bounded queue (maxsize=3)
  producer total menunggu  :  300.6 ms
  kedalaman antrean maks   : 3 pesan
  pesan terakhir menunggu  :  200.3 ms sebelum diproses
  durasi total             :  501.4 ms

Skenario B: unbounded queue (maxsize=0)
  producer total menunggu  :    0.0 ms
  kedalaman antrean maks   : 10 pesan
  pesan terakhir menunggu  :  501.1 ms sebelum diproses
  durasi total             :  501.4 ms

Mari kita telaah baris-baris kode di atas:

Baris 17-24 (producer): q.put(pesan) adalah titik backpressure — pemanggilan tersebut memblokir selama antrean penuh, dan lamanya blocking diakumulasi di stat["tunggu_producer"].
Baris 23 (kirim_pada[pesan] = ...): waktu pengiriman tiap pesan dicatat untuk mengukur berapa lama pesan terakhir menunggu sampai diproses.
Baris 26-32 (consumer): time.sleep(0.05) memainkan pemrosesan lambat; sepuluh pesan berarti konsumen butuh sekitar 500 ms untuk menyelesaikan semuanya.
Baris 40 (q.put(None)): pesan khusus None menjadi sinyal berhenti — consumer keluar dari loop setelah antrean dikosongkan.
Baris 52-54 (main): dua skenario berjalan atas parameter yang sama, satu perbedaan saja: maxsize=3 berbanding maxsize=0.

Membaca Hasil: Memilih Kedalaman Antrean #

Skenario A menunjukkan backpressure bekerja: producer total menunggu 300,6 ms karena antrean berkapasitas tiga terus penuh, dan kedalaman antrean tidak pernah lewat dari tiga pesan. Sebagai imbalannya, pesan terakhir hanya menunggu 200,3 ms sebelum diproses — datanya masih segar. Skenario B membebaskan producer sepenuhnya (0,0 ms menunggu), tetapi seluruh sepuluh pesan menumpuk seketika dan pesan terakhir baru diproses 501,1 ms setelah dikirim; pada kasus deteksi penipuan, keterlambatan setengah detik itu bisa berarti keputusan yang telat.

Perhatikan angka durasi total: kedua skenario sama-sama selesai sekitar 501 ms. Antrean tidak membuat pemrosesan lebih cepat — antrean hanya memindahkan tunggukan dan menentukan siapa yang menunggu. Angka-angka tersebut juga bergeser tipis antar eksekusi (orde sepersepuluh milidetik) karena penjadwalan thread, meski polanya stabil: antrean berbatas menahan producer, antrean bebas membuat pesan menunggu.

Kelas queue.Queue yang kita pakai berasal dari dokumentasi Python dan berlaku untuk thread dalam satu proses. Kebutuhan TokoKita yang sebenarnya — antrean yang bertahan saat proses berhenti, bisa dibaca banyak consumer, dan menampung jutaan pesan — menuntut broker terdistribusi. Broker paling umum untuk kebutuhan tersebut adalah Apache Kafka, bintang utama Chapter Apache Kafka.

Peran Antrian Pesan dalam Sistem Data #

Point-to-Point versus Publish-Subscribe #

Backpressure saat Consumer Lebih Lambat #

Membaca Hasil: Memilih Kedalaman Antrean #