Ingestion dari Message Queue

Saat toko online mengadakan flash sale, layanan pesanan bisa menerima ratusan event baru setiap detik: pesanan dibuat, dibayar, lalu dikirim hampir bersamaan. Jika layanan pesanan memanggil pipeline ingestion secara langsung, setiap lonjakan trafik ikut menekan pipeline, dan setiap gangguan pada pipeline ikut menggagalkan layanan pesanan. Message queue memutus keterikatan langsung tersebut: producer hanya menaruh event ke antrean, sedangkan consumer mengambil event dengan kecepatan yang mampu dicerna. Pada chapter ini kita akan membedah konsep queue, pola consumer loop, acknowledgment, semantik at-least-once, lalu mempraktikkan semuanya dengan queue tiruan berbasis queue.Queue.

Memutus Keterikatan Producer dan Consumer #

Message queue adalah komponen perantara yang menyimpan pesan sementara di antara dua sistem. Producer (misalnya layanan pesanan toko) menulis event ke antrean tanpa menunggu siapa pun memproses event tersebut. Consumer (misalnya pipeline ingestion kita) membaca event satu per satu sesuai kapasitas. Keduanya tidak saling memanggil, sehingga keduanya bisa berjalan, berhenti, dan diskalakan secara terpisah.

Pola Consumer Loop dan Acknowledgment #

Consumer hampir selalu ditulis sebagai consumer loop: satu perulangan tanpa akhir yang mengambil pesan, memproses, lalu meng-acknowledge. Acknowledgment (ack) adalah penanda bahwa pesan sudah selesai diproses dan aman dihapus dari antrean. Jika consumer berhenti sebelum ack terkirim, message queue menganggap pesan belum diproses dan mengirim ulang pesan tersebut ke consumer lain. Mekanisme inilah yang menjadikan queue andal: pesan tidak dibuang begitu consumer mengalami gangguan.

Contoh: Queue Tiruan dengan queue.Queue #

Modul queue dari pustaka standar Python menyediakan queue.Queue, antrean thread-safe yang bisa kita pakai sebagai miniatur message queue dalam satu proses. Kita akan meniru toko online: thread producer mengirim tiga pesanan, thread consumer memproses pesanan satu per satu.

import queue
import threading
import time

antrian_pesanan = queue.Queue()


def produser():
    for pesanan_id in (101, 102, 103):
        print(f"[produser] kirim pesanan #{pesanan_id}")
        antrian_pesanan.put(pesanan_id)
        time.sleep(0.1)
    antrian_pesanan.put(None)  # penanda berhenti


def konsumen():
    while True:
        pesanan = antrian_pesanan.get()
        if pesanan is None:
            antrian_pesanan.task_done()
            break
        print(f"[konsumen] proses pesanan #{pesanan}")
        antrian_pesanan.task_done()


t_produser = threading.Thread(target=produser)
t_konsumen = threading.Thread(target=konsumen)
t_produser.start()
t_konsumen.start()
t_produser.join()
t_konsumen.join()
print("semua pesanan selesai diproses")

Output yang Dihasilkan:

[produser] kirim pesanan #101
[konsumen] proses pesanan #101
[produser] kirim pesanan #102
[konsumen] proses pesanan #102
[produser] kirim pesanan #103
[konsumen] proses pesanan #103
semua pesanan selesai diproses

Mari kita telaah baris-baris kode di atas:

1. Baris 5 (antrian_pesanan = queue.Queue()): membuat antrean thread-safe. Semua operasi put() dan get() aman dipanggil dari thread berbeda tanpa perlu lock manual.
2. Baris 11 (antrian_pesanan.put(pesanan_id)): producer menaruh pesanan ke antrean. Method put() tidak menunggu consumer, sehingga producer bisa terus menerima pesanan dari pengguna.
3. Baris 13 (antrian_pesanan.put(None)): nilai None berperan sebagai sentinel, penanda bahwa tidak ada lagi pesanan dan consumer boleh berhenti.
4. Baris 18 (antrian_pesanan.get()): consumer mengambil satu pesanan. Method get() memblokir thread sampai pesan tersedia, sehingga loop tidak sibuk berputar kosong.
5. Baris 19-21 (if pesanan is None): consumer mengenali sentinel, meng-ack, lalu keluar dari loop dengan break.
6. Baris 23 (antrian_pesanan.task_done()): acknowledgment versi queue.Queue. Setiap get() harus diimbangi satu task_done() agar jumlah pekerjaan tercatat seimbang.

Semantik At-Least-Once #

Queue umumnya menjanjikan semantik at-least-once: setiap pesan diproses minimal satu kali, tetapi boleh lebih dari satu kali jika pesan dikirim ulang setelah consumer gagal. Konsekuensinya, consumer kita harus aman diproses berulang — topik yang akan kita tuntaskan pada Chapter Idempotent Ingestion. Contoh berikut meniru kegagalan tersebut: pesanan 202 gagal pada percobaan pertama, tidak di-ack, lalu dikirim ulang dan berhasil pada percobaan kedua.

import queue
import threading

antrian = queue.Queue()


def produser():
    for pesanan_id in (201, 202, 203):
        antrian.put({"id": pesanan_id, "percobaan": 1})


def konsumen():
    while True:
        pesan = antrian.get()
        if pesan is None:
            antrian.task_done()
            break
        if pesan["id"] == 202 and pesan["percobaan"] == 1:
            print(f"[konsumen] pesanan #{pesan['id']} gagal -> tidak di-ack, kirim ulang")
            antrian.task_done()
            antrian.put({"id": pesan["id"], "percobaan": pesan["percobaan"] + 1})
        else:
            print(f"[konsumen] pesanan #{pesan['id']} selesai (percobaan {pesan['percobaan']}) -> ack")
            antrian.task_done()


t_produser = threading.Thread(target=produser)
t_konsumen = threading.Thread(target=konsumen)
t_produser.start()
t_konsumen.start()
t_produser.join()
antrian.join()         # tunggu semua pesanan benar-benar selesai
antrian.put(None)      # konsumen boleh berhenti
t_konsumen.join()

Output yang Dihasilkan:

[konsumen] pesanan #201 selesai (percobaan 1) -> ack
[konsumen] pesanan #202 gagal -> tidak di-ack, kirim ulang
[konsumen] pesanan #203 selesai (percobaan 1) -> ack
[konsumen] pesanan #202 selesai (percobaan 2) -> ack

Baris 18-21 pada blok kode di atas adalah inti simulasi kegagalan: pesanan 202 dianggap gagal pada percobaan pertama, sehingga consumer memasukkan kembali pesanan tersebut ke antrean dengan penghitung percobaan bertambah. Baris antrian.join() menunggu hingga setiap pesanan yang pernah dimasukkan sudah di-ack, termasuk hasil pengiriman ulang; hanya setelah baris tersebut selesai, sentinel None dikirim agar consumer berhenti dengan rapi. Perhatikan bahwa pesanan 202 tampil dua kali pada output — persis perilaku at-least-once di sistem produksi.

Kafka sebagai Queue Terdistribusi #

Di produksi, message queue berjalan sebagai sistem terdistribusi terpisah. Apache Kafka adalah pilihan paling umum untuk pipeline data: producer menulis event ke topic, consumer membaca topic tersebut sambil mencatat posisi baca bernama offset, dan sekumpulan consumer bisa membaca topic yang sama sebagai consumer group. Semua konsep yang kita praktikkan di atas — consumer loop, ack, pengiriman ulang — tetap berlaku, hanya perantaranya bukan lagi queue.Queue. Kita akan membedah Kafka secara menyeluruh pada Section Streaming Data. Dokumentasi queue.Queue untuk eksplorasi lebih lanjut tersedia di https://docs.python.org/3/library/queue.html .

Antrean event sangat cocok mengalirkan aktivitas yang terjadi saat ini. Namun bagaimana jika sumber data kita adalah isi database yang terus berubah — harga produk diperbarui, pesanan dibatalkan, pelanggan baru terdaftar? Membaca seluruh tabel berulang kali jelas boros. Jawaban modern untuk masalah tersebut adalah Change Data Capture, yang kita bahas pada chapter berikutnya.