KodelokaConsumer Group memungkinkan beberapa worker membagi entri dari satu Stream tanpa tumpang tindih, sambil melacak entri mana yang sudah dikonfirmasi selesai. Entri yang sudah dikirim tapi belum di-XACK masuk ke daftar PEL (Pending Entries List). Bila worker mati, entri tersebut bisa diambil alih worker lain lewat XAUTOCLAIM. Model jaminannya at-least-once, sehingga consumer WAJIB idempoten.
1. Tiga Peran yang Perlu Dibedakan
- Stream — lognya sendiri, misal
pesanan. - Consumer Group — sekelompok worker yang bersama-sama menghabiskan satu stream. Group menyimpan posisi baca terakhir di sisi server, bukan di aplikasi.
- Consumer — anggota individual di dalam group, punya nama sendiri (misal
worker-1).
2. Membuat Group
# Buat group yang mulai membaca dari entri PALING AWAL XGROUP CREATE pesanan grup-notifikasi 0 # Buat group yang hanya membaca entri BARU (setelah perintah ini) XGROUP CREATE pesanan grup-analitik $ # MKSTREAM: buat stream-nya sekalian bila belum ada. # Tanpa ini, perintah gagal kalau stream belum pernah ditulisi. XGROUP CREATE pesanan grup-invoice 0 MKSTREAM # Lihat semua group pada satu stream XINFO GROUPS pesanan
3. Membaca sebagai Anggota Group: XREADGROUP
# Tanda > berarti "beri saya entri yang BELUM PERNAH dikirim # ke consumer mana pun di group ini" XREADGROUP GROUP grup-notifikasi worker-1 COUNT 10 BLOCK 5000 STREAMS pesanan > # ID 0 berarti "beri saya entri yang SUDAH dikirim ke SAYA # tapi belum saya konfirmasi" -- dipakai saat worker baru restart XREADGROUP GROUP grup-notifikasi worker-1 COUNT 10 STREAMS pesanan 0
Saat worker menyala, jangan langsung membaca >. Baca dulu dengan ID 0 untuk menghabiskan pekerjaan yang tertinggal dari sesi sebelumnya, baru beralih ke > setelah daftarnya kosong. Melewatkan langkah ini membuat pesan menggantung di PEL tanpa ada yang mengerjakannya.
4. PEL: Pending Entries List
Begitu sebuah entri dikirim ke consumer lewat XREADGROUP, Redis mencatatnya di PEL — daftar "sedang dikerjakan, belum dikonfirmasi". Entri keluar dari PEL hanya ketika di-XACK.
# Ringkasan: berapa banyak yang menggantung, ID terkecil & terbesar, per consumer XPENDING pesanan grup-notifikasi # Rincian 10 entri menggantung, lengkap dengan # berapa lama menggantung (idle time, ms) dan berapa kali dikirim ulang XPENDING pesanan grup-notifikasi - + 10 # Hanya milik worker tertentu XPENDING pesanan grup-notifikasi - + 10 worker-1 # Konfirmasi selesai -- entri dibuang dari PEL XACK pesanan grup-notifikasi 1712345678901-0
5. Worker Mati: Mengambil Alih dengan XAUTOCLAIM
Kalau worker-1 mati setelah menerima entri tapi sebelum XACK, entri itu terjebak di PEL milik worker-1 selamanya. Worker lain harus mengambil alih.
# Ambil alih entri yang sudah menganggur lebih dari 60 detik (60000 ms) # ke consumer bernama worker-2, mulai dari kursor 0-0 XAUTOCLAIM pesanan grup-notifikasi worker-2 60000 0-0 COUNT 10 # Hasilnya: [kursor-berikutnya, [entri...], [id-yang-sudah-dihapus...]] # Panggil berulang dengan kursor yang dikembalikan sampai kursornya 0-0. # XCLAIM adalah versi manual untuk ID spesifik XCLAIM pesanan grup-notifikasi worker-2 60000 1712345678901-0
Kalau sebuah entri selalu membuat worker crash, entri itu akan di-claim berulang kali tanpa henti dan memblokir kemajuan group. Periksa kolom delivery count dari XPENDING: bila sudah melewati ambang tertentu (misal 5 kali), pindahkan entri itu ke stream pesanan-dlq lalu XACK aslinya. Ini setara Dead Letter Queue di RabbitMQ, hanya saja di Redis Anda harus membangunnya sendiri.
6. Jaminan Pengiriman: At-Least-Once, Bukan Exactly-Once
Urutan kejadiannya begini:
1. XREADGROUP -> entri masuk PEL 2. Worker memproses (misal: kirim email) 3. XACK -> entri keluar dari PEL
Kalau worker mati di antara langkah 2 dan 3, email sudah terkirim tetapi entri belum di-ACK. Worker lain akan mengambil alih dan mengirim email itu lagi.
Yang bisa Anda capai adalah efek exactly-once melalui idempotensi: simpan ID entri yang sudah selesai diproses (misal di Set Redis atau kolom unik di database), dan lewati bila sudah pernah ada. Cara paling kokoh: jadikan penulisan hasil dan penandaan "sudah diproses" berada dalam satu transaksi database yang sama.
7. Membersihkan Consumer yang Sudah Tidak Dipakai
# Lihat consumer di dalam group beserta idle time-nya XINFO CONSUMERS pesanan grup-notifikasi # Hapus consumer yang sudah mati permanen (misal pod lama di Kubernetes). # PERINGATAN: entri di PEL miliknya ikut hilang -- claim dulu sebelum menghapus! XGROUP DELCONSUMER pesanan grup-notifikasi worker-lama # Hapus group seluruhnya XGROUP DESTROY pesanan grup-analitik # Geser posisi baca group secara manual (misal untuk replay dari awal) XGROUP SETID pesanan grup-analitik 0
Kalau nama consumer memakai hostname pod yang berubah setiap deploy, PEL akan penuh dengan consumer yatim. Pakai StatefulSet (nama pod stabil: worker-0, worker-1) atau jalankan XAUTOCLAIM secara berkala sebagai jaring pengaman.