Ketika konsumen tertinggal dari produsen, pesan menumpuk dan latensi pemrosesan melonjak. Offset yang dikonfigurasi salah dapat diam-diam melewatkan atau memproses ulang pesan. Panduan ini mencakup pola-pola terbukti untuk membangun aplikasi konsumen yang andal dan ber-throughput tinggi di ApsaraMQ for Kafka, mencakup kelompok konsumen, manajemen offset, penanganan kegagalan, dan penyetelan performa.
Cara kerja konsumsi pesan
Konsumen ApsaraMQ for Kafka mengulangi siklus tiga langkah berikut:
Poll — Ambil batch pesan dari broker.
Process — Jalankan logika bisnis Anda pada pesan yang diambil.
Poll again — Ambil batch berikutnya setelah pemrosesan selesai.
Jaga waktu pemrosesan tetap singkat agar interval poll tetap stabil. Penundaan pemrosesan yang lama dapat memicu rebalance atau menyebabkan akumulasi pesan.
Kelompok konsumen dan penyeimbangan beban
Kelompok konsumen adalah kumpulan instans konsumen yang memiliki group.id yang sama. ApsaraMQ for Kafka mendistribusikan partisi dari topik yang dilanggan secara merata ke seluruh konsumen dalam satu kelompok, sehingga setiap pesan dikirimkan tepat ke satu konsumen.
Contoh: Kelompok A berlangganan ke Topik A. Tiga konsumen — C1, C2, dan C3 — berada dalam kelompok tersebut. Setiap pesan masuk akan dikirim ke salah satu dari C1, C2, atau C3, sehingga beban konsumsi tersebar merata di ketiganya.
ApsaraMQ for Kafka memicu rebalance saat konsumen:
Pertama kali dijalankan
Dijalankan ulang
Ditambahkan ke atau dihapus dari kelompok
Cegah rebalance yang sering
Rebalance menghentikan sementara konsumsi selama partisi dialokasikan ulang. Rebalance yang sering mengganggu throughput dan meningkatkan latensi pemrosesan. Hal ini terjadi ketika heartbeat konsumen mengalami timeout.
Untuk mencegah rebalance yang sering, sesuaikan parameter terkait atau tingkatkan laju konsumsi. Untuk langkah troubleshooting detail, lihat Mengapa rebalance sering terjadi pada client konsumen saya?
Konfigurasi partisi
Jumlah partisi menentukan berapa banyak konsumen yang dapat memproses pesan secara paralel. Setiap partisi ditugaskan ke tepat satu konsumen dalam kelompok — jika jumlah konsumen melebihi jumlah partisi, konsumen tambahan akan menganggur.
Jumlah partisi yang direkomendasikan
Jumlah partisi default di Konsol ApsaraMQ for Kafka adalah 12, yang cocok untuk sebagian besar beban kerja. Saat melakukan scaling, ikuti panduan berikut:
| Jumlah partisi | Dampak |
|---|---|
| Kurang dari 12 | Dapat mengurangi performa produksi dan konsumsi pesan |
| 12 – 100 | Rentang yang direkomendasikan untuk sebagian besar beban kerja |
| Lebih dari 100 | Dapat memicu rebalance konsumen yang sering |
Anda tidak dapat mengurangi jumlah partisi setelah menambahkannya. Tingkatkan jumlah partisi secara bertahap, bukan dalam lompatan besar.
Pola langganan
ApsaraMQ for Kafka mendukung dua pola langganan.
Satu kelompok berlangganan ke beberapa topik
Satu kelompok konsumen dapat berlangganan ke beberapa topik. Pesan dari semua topik yang dilanggan didistribusikan secara merata ke seluruh konsumen dalam kelompok tersebut.
String topicStr = kafkaProperties.getProperty("topic");
String[] topics = topicStr.split(",");
for (String topic: topics) {
subscribedTopics.add(topic.trim());
}
consumer.subscribe(subscribedTopics);Beberapa kelompok berlangganan ke satu topik
Beberapa kelompok konsumen dapat secara independen berlangganan ke topik yang sama. Setiap kelompok menerima semua pesan, dan kelompok-kelompok tersebut beroperasi secara independen tanpa saling memengaruhi.
Gunakan pola ini ketika aplikasi berbeda perlu memproses aliran data yang sama secara independen — misalnya, satu kelompok untuk analitik real-time dan kelompok lain untuk pengarsipan data.
Gunakan satu kelompok per aplikasi
Gunakan kelompok konsumen khusus untuk setiap aplikasi. Jika satu aplikasi perlu menjalankan logika konsumsi yang berbeda, buat file konfigurasi terpisah (misalnya, kafka1.properties dan kafka2.properties) dengan nilai group.id yang berbeda.
Sesuaikan langganan dalam satu kelompok
Semua konsumen dalam kelompok yang sama harus berlangganan ke himpunan topik yang sama. Langganan yang tidak sesuai mempersulit troubleshooting dan dapat menyebabkan perilaku rebalance yang tidak terduga.
Kelola offset konsumen
Setiap partisi melacak offset maksimum — jumlah total pesan yang diterima. Setiap konsumen melacak offset konsumen — jumlah pesan yang telah diproses dalam partisi tersebut. Selisih antara keduanya adalah akumulasi pesan (backlog yang belum dikonsumsi).
Auto-commit vs commit manual
ApsaraMQ for Kafka menyediakan dua parameter untuk melakukan commit offset konsumen:
| Parameter | Default | Deskripsi |
|---|---|---|
enable.auto.commit | true | Mengaktifkan commit offset otomatis |
auto.commit.interval.ms | 1000 (ms) | Interval antar commit otomatis |
Dengan auto-commit diaktifkan, client memeriksa waktu sejak commit terakhir sebelum setiap poll. Jika waktu yang berlalu melebihi auto.commit.interval.ms, client akan melakukan commit offset saat ini.
Saat menggunakan auto-commit, pastikan semua pesan dari poll sebelumnya telah sepenuhnya diproses sebelum poll berikutnya. Jika konsumen menulis ke datastore eksternal dan penulisan tersebut gagal, auto-commit tetap dapat memajukan offset — menyebabkan pesan tersebut dilewati secara permanen tanpa dicoba ulang. Pahami pertukaran ini sebelum mengandalkan perilaku default.
Lakukan commit offset secara manual
Untuk melakukan commit offset secara manual, atur enable.auto.commit ke false dan panggil fungsi commit(offsets) setelah memproses setiap batch.
Perilaku reset offset
Offset konsumen di-reset dalam dua situasi:
Tidak ada offset yang telah di-commit — misalnya, saat konsumen terhubung ke broker untuk pertama kalinya.
Offset yang di-commit tidak valid — misalnya, offset maksimum dalam partisi adalah 10, tetapi konsumen mencoba membaca dari offset 11.
Kendalikan perilaku reset pada client Java dengan auto.offset.reset:
| Nilai | Perilaku |
|---|---|
latest | Mulai dari pesan terbaru. Gunakan ini sebagai default Anda |
earliest | Mulai dari pesan tertua yang tersedia |
none | Lempar exception alih-alih melakukan reset |
Lebih baik gunakan latest daripada earliest untuk menghindari pemrosesan ulang seluruh riwayat topik saat terjadi offset tidak valid. Jika Anda melakukan commit offset secara manual dengan fungsi commit(offsets), none merupakan pilihan aman karena offset Anda seharusnya selalu valid.
Menangani pesan besar
Saat ukuran pesan individu melebihi ukuran biasa, sesuaikan parameter berikut untuk mengontrol perilaku fetch:
| Parameter | Panduan |
|---|---|
max.poll.records | Jumlah maksimum record per poll. Atur ke 1 untuk pesan yang lebih besar dari 1 MB |
fetch.max.bytes | Jumlah data maksimum per permintaan fetch. Atur sedikit lebih besar dari ukuran pesan yang diharapkan |
max.partition.fetch.bytes | Jumlah data maksimum per partisi per fetch. Atur sedikit lebih besar dari ukuran pesan yang diharapkan |
Dengan pengaturan ini, konsumen mengambil pesan besar satu per satu, mencegah tekanan memori akibat batch yang terlalu besar.
Menangani duplikasi pesan
ApsaraMQ for Kafka menggunakan semantik pengiriman at-least-once: setiap pesan dikirimkan minimal sekali untuk mencegah kehilangan data, tetapi duplikasi dapat terjadi selama error jaringan atau restart client.
Implementasikan konsumsi idempoten
Jika aplikasi Anda sensitif terhadap duplikasi — misalnya, transaksi finansial atau pemrosesan pesanan — implementasikan deduplikasi di tingkat aplikasi:
Tambahkan kunci unik ke setiap pesan saat memproduksi (misalnya, ID transaksi).
Sebelum memproses pesan yang dikonsumsi, periksa apakah kunci tersebut sudah pernah diproses.
Lewati pesan dengan kunci yang sudah pernah diproses.
Jika aplikasi Anda dapat mentolerir duplikasi sesekali (misalnya, agregasi metrik atau ingestion log), lewati langkah ini.
Menangani kegagalan konsumsi
Pesan dalam satu partisi dikonsumsi secara berurutan. Saat pemrosesan gagal — misalnya, karena data rusak atau gangguan layanan downstream — pilih salah satu strategi berikut:
| Strategi | Trade-off |
|---|---|
| Retry in place | Coba ulang pesan yang gagal hingga berhasil. Mudah diimplementasikan, tetapi memblokir thread konsumen dan menyebabkan akumulasi pesan pada partisi tersebut |
| Dead-letter topic | Kirim pesan yang gagal ke topik khusus untuk inspeksi lanjutan. Konsumsi berlanjut tanpa pemblokiran, tetapi memerlukan proses terpisah untuk menyelidiki dan memproses ulang kegagalan |
Untuk sebagian besar beban kerja produksi, pendekatan dead-letter menghindari penghentian pipeline konsumen.
Mengurangi latensi konsumsi
ApsaraMQ for Kafka menggunakan model berbasis pull: konsumen mengambil pesan dari broker pada setiap siklus poll. Saat konsumen mampu mengimbangi produsen, latensi tetap rendah. Lonjakan latensi biasanya mengindikasikan akumulasi pesan.
Penyebab umum akumulasi
| Penyebab | Solusi |
|---|---|
| Laju konsumsi lebih lambat daripada laju produksi | Tambahkan konsumen atau tingkatkan throughput pemrosesan |
| Thread konsumen diblokir oleh panggilan remote yang lambat | Atur timeout pada panggilan eksternal agar thread dilepas setelah menunggu dalam batas waktu tertentu |
Tingkatkan throughput konsumsi
Tambahkan lebih banyak konsumen. Jalankan instans konsumen tambahan (satu thread per konsumen) atau deploy lebih banyak proses konsumen. Jumlah konsumen aktif tidak boleh melebihi jumlah partisi — konsumen tambahan akan menganggur.
Gunakan kolam thread untuk pemrosesan. Pisahkan polling dari pemrosesan agar pekerjaan dapat diparalelkan:
Definisikan kolam thread dengan jumlah tetap thread pekerja.
Ambil batch pesan.
Kirimkan setiap pesan ke kolam thread untuk diproses secara konkuren.
Setelah semua tugas dalam batch selesai, ambil batch berikutnya.
Pola ini menjaga loop poll tetap responsif sambil mendistribusikan beban pemrosesan ke beberapa thread.
Penyaringan pesan
ApsaraMQ for Kafka tidak menyediakan penyaringan pesan bawaan. Implementasikan penyaringan di tingkat aplikasi menggunakan salah satu pendekatan berikut:
| Pendekatan | Kapan digunakan |
|---|---|
| Topik terpisah | Jumlah kecil kategori pesan yang berbeda |
| Penyaringan di sisi client | Banyak kategori, atau logika penyaringan yang sering berubah |
Gabungkan kedua pendekatan ini jika kasus penggunaan Anda memerlukannya — arahkan kategori luas ke topik berbeda, lalu terapkan penyaringan detail halus di sisi konsumen.
Penyiaran pesan
ApsaraMQ for Kafka tidak mendukung penyiaran pesan secara native. Untuk mengirimkan setiap pesan ke beberapa konsumen independen, buat kelompok konsumen terpisah untuk setiap konsumen yang perlu menerima semua pesan. Setiap kelompok secara independen mengonsumsi seluruh aliran pesan dari topik tersebut.