Consumer group

Consumer group menyelesaikan persoalan yang tidak dijawab XREAD biasa, yaitu pembagian pesan kepada beberapa pekerja sekaligus pencatatan pesan yang belum selesai diproses. Topik ini membahas pembuatan group dengan XGROUP, penyerahan pesan dengan XREADGROUP, konfirmasi pemrosesan dengan XACK, daftar pending yang mencatat pesan yang belum di-ack, serta pemulihan pesan macet dengan XCLAIM dan XAUTOCLAIM. Setelah mengikuti contoh pada topik ini, antrean tugas dengan jaminan pengiriman at-least-once, yaitu setiap pesan diproses paling sedikit satu kali, dapat dibangun di atas Redis.

Membuat consumer group dengan XGROUP#

Consumer group dicatat pada stream melalui XGROUP CREATE dengan tiga informasi utama, yaitu nama key, nama group, dan posisi ID tempat group mulai membaca. Posisi 0 berarti group mengambil seluruh entri yang sudah ada, sedangkan posisi $ berarti group hanya melihat entri yang ditambahkan setelahnya. Satu stream boleh memuat lebih dari satu group, dan setiap group menyimpan posisi baca serta daftar pending miliknya sendiri. Argumen MKSTREAM meminta stream dibuat sekaligus apabila key-nya belum ada, sehingga group dapat disiapkan lebih dahulu sebelum entri pertama masuk.

# menghapus key lama agar contoh selalu dimulai dari keadaan yang sama, keluarannya diabaikan
redis-cli del antrean:cetak >/dev/null
# group dibuat dari posisi awal, MKSTREAM membuat streamnya sekaligus
redis-cli xgroup create antrean:cetak prosesor 0 mkstream
# pembuatan group dengan nama yang sama ditolak
redis-cli xgroup create antrean:cetak prosesor 0 mkstream
# empat tugas cetak masuk ke stream, keluaran penambahannya diabaikan
redis-cli xadd antrean:cetak 1-1 dokumen skripsi-ti halaman 84 >/dev/null
redis-cli xadd antrean:cetak 1-2 dokumen makalah-sejarah halaman 12 >/dev/null
redis-cli xadd antrean:cetak 2-1 dokumen modul-praktikum halaman 40 >/dev/null
redis-cli xadd antrean:cetak 2-2 dokumen jadwal-ujian halaman 2 >/dev/null

OK
BUSYGROUP Consumer Group name already exists

Pembuatan pertama mengembalikan OK, sedangkan pembuatan kedua dengan nama yang sama ditolak dengan pesan error BUSYGROUP. Penolakan ini terjadi karena nama group bersifat unik per stream, dan posisi baca milik group yang sudah ada tidak boleh ditimpa oleh pembuatan yang baru. Apabila keadaan group memang perlu direset, XGROUP DESTROY menghapus group beserta daftar pendingnya, lalu group yang baru dapat dibuat kembali dari posisi yang diinginkan.

Menyerahkan pesan dengan XREADGROUP#

XREADGROUP membaca pesan atas nama satu consumer di dalam group dengan menyebut nama group dan nama consumer sebelum bagian STREAMS. Tanda > pada posisi ID berarti serahkan pesan yang belum pernah diserahkan kepada consumer mana pun, dan pesan yang diserahkan langsung tercatat pada daftar pending group atas nama consumer penerimanya. Dua consumer yang membaca group yang sama menerima pesan yang berbeda karena setiap pesan hanya diserahkan kepada satu consumer. Sifat ini membedakan XREADGROUP dari XREAD biasa, yang membaca seluruh isi stream secara independen tanpa pembagian pesan.

Diagram berikut merangkum alur lengkapnya, yaitu entri masuk ke stream, group mendistribusikannya kepada consumer yang meminta, entri yang sudah diserahkan masuk daftar pending, dan XACK menghapusnya dari daftar setelah pemrosesan berhasil.

# pembaca-a mengambil dua tugas yang belum pernah diserahkan
redis-cli xreadgroup group prosesor pembaca-a count 2 streams antrean:cetak '>'
# pembaca-b mengambil dua tugas berikutnya
redis-cli xreadgroup group prosesor pembaca-b count 2 streams antrean:cetak '>'

antrean:cetak
1-1
dokumen
skripsi-ti
halaman
84
1-2
dokumen
makalah-sejarah
halaman
12
antrean:cetak
2-1
dokumen
modul-praktikum
halaman
40
2-2
dokumen
jadwal-ujian
halaman
2

Pembaca-a menerima 1-1 dan 1-2, sedangkan pembaca-b menerima 2-1 dan 2-2, tanpa satu pun pesan yang diserahkan dua kali. Argumen COUNT membatasi jumlah pesan per pemanggilan, sehingga consumer memproses sekumpulan kecil pesan sebelum meminta lagi. Apabila tidak ada pesan baru yang tersisa, XREADGROUP mengembalikan keluaran kosong, dan bentuk BLOCK tersedia untuk menunggu pesan berikutnya sebagaimana XREAD.

Daftar pending dan XACK#

Setiap group menyimpan daftar pending, yaitu catatan pesan yang sudah diserahkan kepada consumer tetapi belum dikonfirmasi selesai. Pesan masuk daftar pada saat diserahkan dan keluar hanya melalui XACK, sehingga daftar ini menjadi sumber kebenaran tentang pekerjaan yang sedang berjalan. XPENDING meringkas isinya, yaitu jumlah pesan, rentang ID, dan pembagian per consumer, sedangkan XACK menghapus pesan dari daftar setelah pemrosesannya dikonfirmasi.

# ringkasan daftar pending group prosesor
redis-cli xpending antrean:cetak prosesor
# pembaca-a menyelesaikan dua tugasnya
redis-cli xack antrean:cetak prosesor 1-1 1-2
# ringkasan daftar pending setelah ack
redis-cli xpending antrean:cetak prosesor

4
1-1
2-2
pembaca-a
2
pembaca-b
2
2
2
2-1
2-2
pembaca-b
2

Sebelum ack, daftar memuat empat pesan yang terbagi rata kepada dua consumer. XACK 1-1 1-2 mengembalikan angka dua, yaitu jumlah pesan yang berhasil dihapus dari daftar, dan ringkasan berikutnya hanya memuat dua pesan milik pembaca-b. Pesan yang tertinggal di daftar bukanlah kegagalan, melainkan tanda pemrosesan belum dikonfirmasi, dan catatan inilah yang dimanfaatkan untuk memulihkan pesan macet.

Memulihkan pesan macet dengan XCLAIM dan XAUTOCLAIM#

Pesan di dalam daftar pending menjadi macet apabila consumer pemegangnya berhenti sebelum meng-ack. XCLAIM memindahkan pesan tertentu kepada consumer lain dengan menyebut ID-nya, sedangkan XAUTOCLAIM memindahkan pesan yang waktu diamnya melampaui ambang tanpa perlu menyebut ID satu per satu. Keduanya menerima ambang waktu diam dalam milidetik, dan pesan hanya dipindahkan apabila diamnya melewati ambang tersebut. Setiap pemindahan menaikkan penghitung penyerahan pesan, sehingga pesan yang berulang kali gagal dapat dikenali dan diperiksa penyebabnya.

# pesan 2-1 dipegang pembaca-b yang berhenti, diambil alih pembaca-a dengan ambang diam 0
redis-cli xclaim antrean:cetak prosesor pembaca-a 0 2-1
# pembagian pesan pending setelah pengambilalihan
redis-cli xpending antrean:cetak prosesor

2-1
dokumen
modul-praktikum
halaman
40
2
2-1
2-2
pembaca-a
1
pembaca-b
1

Ambang 0 milidetik pada contoh ini dipilih agar pemindahan pasti terjadi, sedangkan produksi menggunakan ambang sesuai durasi tunggu yang wajar bagi pesan, misalnya 30000 untuk tiga puluh detik. XCLAIM mengembalikan isi pesan yang dipindahkan, sehingga consumer baru langsung dapat memprosesnya. Ringkasan daftar pending memperlihatkan pembaca-a dan pembaca-b kini masing-masing memegang satu pesan.

# seluruh pesan yang diamnya melewati ambang 0 dipindahkan kepada pembaca-a
redis-cli xautoclaim antrean:cetak prosesor pembaca-a 0 0-0 count 10
# seluruh pesan pending kini dipegang pembaca-a
redis-cli xpending antrean:cetak prosesor
# kedua pesan selesai diproses dan di-ack
redis-cli xack antrean:cetak prosesor 2-1 2-2
# daftar pending kosong setelah ack
redis-cli xpending antrean:cetak prosesor

0-0
2-1
dokumen
modul-praktikum
halaman
40
2-2
dokumen
jadwal-ujian
halaman
2

2
2-1
2-2
pembaca-a
2
2
0

XAUTOCLAIM mengembalikan ID kursor untuk pemanggilan berikutnya beserta isi pesan yang berhasil dipindahkan, sehingga pemulihan dapat berjalan bertahap dalam perulangan. Pesan 2-1 ikut dikembalikan kepada pembaca-a meskipun baru saja dipindahkan melalui XCLAIM, karena ambang 0 milidetik tidak menyaring apa pun. Setelah kedua pesan di-ack, ringkasan daftar pending menunjukkan angka nol dan antrean kembali bersih.

Kesalahan umum#

Kesalahan pada consumer group umumnya berupa pembuatan ganda, rujukan group yang sudah dihapus, dan pesan yang menumpuk tanpa ack. Daftar berikut memuat penyebab dan perbaikannya.

• Membuat group dengan nama yang sudah terdaftar. XGROUP CREATE menolaknya dengan error BUSYGROUP karena posisi baca group lama tidak boleh ditimpa. Tangkap error tersebut, atau hapus group lama dengan XGROUP DESTROY terlebih dahulu bila keadaannya memang perlu direset.
• Menjalankan XREADGROUP terhadap group yang sudah dihapus. Perintahnya ditolak dengan error NOGROUP sampai group dibuat kembali, dan pesan errornya menyebut key serta group yang tidak ditemukan sehingga penyebabnya mudah dilacak.
• Menyerahkan pesan tanpa pernah meng-ack. Daftar pending membesar tanpa batas dan pesan yang sama diproses berulang oleh jalur pemulihan. Pastikan setiap jalur eksekusi, termasuk jalur yang gagal, berujung pada XACK atau pemindahan yang tercatat.

Error NOGROUP pada butir kedua ditunjukkan berikut. Penghapusan group mengembalikan jumlah group yang terhapus, dan XREADGROUP terhadap group yang sama langsung ditolak.

# menghapus group prosesor, keluarannya jumlah group yang terhapus
redis-cli xgroup destroy antrean:cetak prosesor
# XREADGROUP terhadap group yang sudah tidak ada
redis-cli xreadgroup group prosesor pembaca-a count 1 streams antrean:cetak '>'

1
NOGROUP No such key 'antrean:cetak' or consumer group 'prosesor' in XREADGROUP with GROUP option

Ringkasan#

• XGROUP CREATE mendaftarkan group pada posisi tertentu dan menolak nama ganda dengan BUSYGROUP, sedangkan XGROUP DESTROY menghapus group beserta daftar pendingnya.
• XREADGROUP dengan tanda > menyerahkan setiap pesan yang belum pernah diserahkan kepada tepat satu consumer, sehingga beberapa consumer dapat membagi beban dengan aman.
• Pesan yang diserahkan masuk daftar pending sampai XACK menghapusnya, dan XPENDING merangkum sisa pekerjaan per consumer.
• Pesan macet dipulihkan dengan XCLAIM untuk ID tertentu atau XAUTOCLAIM untuk seluruh pesan yang diamnya melampaui ambang, dengan penghitung penyerahan sebagai penanda kegagalan berulang.

Referensi#

• XGROUP — Redis — Dokumentasi XGROUP CREATE dan DESTROY beserta semantik posisi awal group.
• XREADGROUP — Redis — Dokumentasi XREADGROUP beserta tanda > serta opsi COUNT dan BLOCK.
• XACK — Redis — Dokumentasi XACK untuk menghapus pesan dari daftar pending.
• XPENDING — Redis — Dokumentasi XPENDING dalam bentuk ringkasan dan rincian per pesan.
• XCLAIM — Redis — Dokumentasi XCLAIM untuk memindahkan pesan pending antar-consumer.
• XAUTOCLAIM — Redis — Dokumentasi XAUTOCLAIM untuk pemulihan pesan macet secara otomatis.