ApsaraMQ for RocketMQ:Penyeimbangan beban konsumen

Cara kerja

Broker mendistribusikan pesan individual dari suatu topik secara merata ke seluruh consumer dalam satu consumer group, terlepas dari antrian tempat pesan tersebut berada. Beberapa consumer dapat memproses pesan dari antrian yang sama secara bersamaan.

Dalam contoh ini, Consumer Group A memiliki tiga consumer: A1, A2, dan A3. Ketiganya mengonsumsi pesan dari Queue1. Broker menetapkan setiap pesan ke satu consumer dalam satu waktu.

Saat sebuah consumer menerima pesan, broker mengunci pesan tersebut sehingga tidak terlihat oleh consumer lain. Pesan tetap terkunci hingga consumer mengonfirmasi penerimaannya atau kunci tersebut kedaluwarsa. Hal ini mencegah pemrosesan duplikat dalam kondisi normal.

Penanganan pesan terurut

Untuk pesan terurut, ApsaraMQ for RocketMQ menjamin bahwa pesan dalam grup pesan yang sama diproses dalam urutan persis seperti saat disimpan di broker.

Pertimbangkan empat pesan terurut (M1 hingga M4) dalam grup pesan G1 dari Queue1. Jika consumer A1 sedang memproses M1 dan M2, consumer A2 tidak dapat mulai memproses M3 atau M4 hingga A1 mengirimkan status konsumsi untuk M1 dan M2. Broker menerapkan penguncian sekuensial yang ketat untuk menjaga urutan.

Keunggulan dibanding penyeimbangan beban berbasis antrian

Kapan digunakan

Penyeimbangan beban berbasis pesan cocok untuk sebagian besar beban kerja pemrosesan event online di mana setiap pesan ditangani secara independen—misalnya, pemrosesan pesanan, pengiriman notifikasi, dan penanganan event real-time.

Untuk beban kerja pemrosesan aliran atau agregasi yang memerlukan pemrosesan batch pesan dari antrian yang sama, gunakan penyeimbangan beban berbasis antrian.

Cakupan

Penyeimbangan beban berbasis pesan adalah satu-satunya kebijakan yang tersedia untuk tipe Push Consumer dan Simple Consumer pada broker versi 5.x. Kebijakan ini diaktifkan secara default dan tidak memerlukan konfigurasi tambahan.

Contoh

Tipe Push Consumer dan Simple Consumer menggunakan penyeimbangan beban berbasis pesan secara otomatis. Contoh Java berikut menunjukkan kedua tipe consumer memproses pesan tanpa konfigurasi penyeimbangan beban apa pun.

Push Consumer dengan pendengar pesan:

// Push Consumer: implementasikan pendengar pesan untuk memproses pesan.
// Penyeimbangan beban ditangani secara otomatis oleh broker.
MessageListener messageListener = new MessageListener() {
    @Override
    public ConsumeResult consume(MessageView messageView) {
        System.out.println(messageView);
        // Kembalikan hasil konsumsi.
        return ConsumeResult.SUCCESS;
    }
};

Konsumen Sederhana dengan Konfirmasi Manual:

// Simple Consumer: tarik pesan, proses, lalu akui setiap pesan.
// Penyeimbangan beban ditangani secara otomatis oleh broker.
try {
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        try {
            // Akui pesan setelah diproses.
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // Jika penarikan gagal karena pembatasan kecepatan atau masalah lain, ulangi permintaan.
    e.printStackTrace();
}

Cara kerja

Broker menetapkan setiap antrian dalam suatu topik ke tepat satu consumer dalam consumer group. Setiap consumer kemudian memproses semua pesan dari antrian yang ditetapkan kepadanya.

Dalam contoh ini, suatu topik memiliki tiga antrian (Queue1, Queue2, Queue3) dan consumer group memiliki dua consumer. Karena setiap antrian ditetapkan ke satu consumer, consumer A2 mendapatkan dua antrian sementara A1 mendapatkan satu. Jika jumlah antrian lebih sedikit daripada jumlah consumer, beberapa consumer tidak menerima antrian dan tetap menganggur.

Setiap consumer mengikuti urutan pemrosesan berikut: tarik pesan dari antrian yang ditetapkan, kirimkan offset konsumsi, dan simpan offset tersebut. Karena status konsumsi tidak dikembalikan ke antrian saat consumer menarik pesan, setiap antrian harus ditetapkan secara eksklusif ke satu consumer untuk mencegah pemrosesan duplikat.

Keunggulan dibanding penyeimbangan beban berbasis pesan

Kapan digunakan

Penyeimbangan beban berbasis antrian paling cocok untuk aplikasi komputasi aliran dan agregasi data yang perlu memproses batch atau mengagregasi pesan dari sumber yang sama. Misalnya, sebuah consumer dapat mengumpulkan metrik dari satu antrian selama jendela waktu tertentu dan menghitung rata-rata berjalan.

Cakupan

Penyeimbangan beban berbasis antrian adalah satu-satunya kebijakan yang tersedia untuk consumer pada broker versi 3.x dan 4.x, termasuk tipe Pull Consumer, Push Consumer bawaan, Pull Consumer bawaan, dan Lite Pull Consumer. Pada broker versi 5.x, tipe Pull Consumer tetap menggunakan penyeimbangan beban berbasis antrian secara default.

Tidak diperlukan konfigurasi tambahan—penyeimbangan beban berbasis antrian diaktifkan secara otomatis untuk tipe-tipe consumer ini.

Contoh

Untuk contoh kode, lihat LitePullConsumerAssign.java dalam pustaka kode Apache RocketMQ.

Implementasikan penanganan pesan idempoten

Kedua kebijakan penyeimbangan beban memicu penyeimbangan ulang sementara saat consumer ditambahkan, dihapus, atau saat broker diskalakan. Selama penyeimbangan ulang, sejumlah kecil pesan mungkin dikirim lebih dari sekali. Untuk menanganinya, implementasikan deduplikasi guna memastikan idempotensi dalam logika konsumsi pesan Anda.