ApsaraMQ for RocketMQ:Jenis konsumen

Cara kerja

SDK menggunakan model thread Reactor secara internal:

1. Thread long-polling bawaan menarik pesan dari server secara asinkron.

2. Pesan ditempatkan ke dalam antrian cache internal.

3. SDK mendistribusikan pesan ke thread konsumen, yang kemudian memanggil listener Anda.

Hasil listener

Listener pesan harus mengembalikan salah satu hasil berikut:

Batasan keandalan

Push consumer menentukan sukses atau gagal secara ketat berdasarkan nilai kembali listener. Untuk menjaga jaminan ini:

1. Proses secara sinkron. Selesaikan seluruh pemrosesan sebelum mengembalikan hasil.

2. Jangan mendistribusikan ulang pesan. Jangan meneruskan pesan ke thread lain dan mengembalikan hasil sebelum thread tersebut selesai.

Jika listener mengembalikan sukses sebelum pemrosesan selesai dan pemrosesan kemudian gagal, server menganggap pesan telah dikonsumsi dan tidak melakukan retry.

Pengiriman pesan terurut

Ketika kelompok konsumen menggunakan mode konsumsi terurut, Push consumer memanggil listener dalam urutan pesan yang ketat tanpa konfigurasi tambahan. Untuk informasi lebih lanjut, lihat Ordered messages.

Kapan menggunakan Push consumers

Push consumers paling efektif ketika:

• Waktu pemrosesan dapat diprediksi. Durasi yang tidak dapat diprediksi sering memicu timeout, yang menyebabkan pesan duplikat melalui mekanisme retry.

• Konsumsi standar sudah cukup. SDK mengontrol model thread dan mengirimkan pesan dengan throughput maksimum. Hal ini menyederhanakan pengembangan tetapi tidak mendukung pemrosesan asinkron atau kontrol laju kustom.

Kelas SDK

ApsaraMQ for RocketMQ menyediakan dua kelas SDK untuk Push consumers:

• PushConsumer — Mengonsumsi pesan dari topik standar (non-Lite).

• LitePushConsumer — Mengonsumsi pesan dari topik Lite, dengan kontrol konsumsi pada tingkat granularitas topik Lite.

// Mengonsumsi pesan normal dengan PushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        // Setel kelompok konsumen
        .setConsumerGroup("<your-consumer-group>")
        // Setel titik akses
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind langganan
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // Daftarkan listener pesan
        .setMessageListener(new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                // Proses pesan dan kembalikan hasil
                return ConsumeResult.SUCCESS;
            }
        })
        .build();

// Mengonsumsi pesan normal dengan LitePushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();

LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
        // Setel titik akses
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Setel topik
        .bindTopic("<your-topic>")
        // Setel kelompok konsumen
        .setConsumerGroup("<your-consumer-group>")
        // Daftarkan listener pesan
        .setMessageListener(messageView -> {
            // Proses pesan dan kembalikan hasil
            return ConsumeResult.SUCCESS;
        })
        .build();

// Berlangganan ke topik Lite
litePushConsumer.subscribeLite("<your-lite-topic-1>");
litePushConsumer.subscribeLite("<your-lite-topic-2>");

Cara kerja

1. Panggil ReceiveMessage untuk menarik batch pesan dari server.

2. Distribusikan pesan ke thread bisnis Anda untuk diproses.

3. Panggil AckMessage untuk setiap pesan yang berhasil diproses.

Jika pemrosesan gagal, jangan kirim acknowledgment. Pesan akan tersedia kembali setelah durasi invisibility pesan berakhir, memicu retry. Untuk informasi lebih lanjut, lihat kebijakan retry SimpleConsumer.

Operasi API

Penanganan kegagalan

Tabel berikut menjelaskan bagaimana skenario kegagalan berbeda memengaruhi pengiriman pesan:

Pengiriman pesan terurut

Simple consumer memproses pesan terurut sesuai urutan penyimpanan. Untuk sekelompok pesan yang harus tetap berurutan, pesan berikutnya tidak dapat diambil hingga pesan sebelumnya selesai diproses.

Kapan menggunakan Simple consumers

Simple consumers paling efektif ketika:

• Waktu pemrosesan tidak dapat diprediksi. Tentukan durasi invisibility pesan awal saat memanggil ReceiveMessage, lalu perpanjang dengan ChangeInvisibleDuration jika diperlukan.

• Diperlukan alur kerja kustom. SDK tidak memaksakan model threading — implementasikan distribusi asinkron, konsumsi batch, atau pola kustom apa pun.

• Kontrol laju penting. Kode Anda menentukan kapan dan seberapa sering memanggil ReceiveMessage, memberikan kendali langsung atas throughput.

Kelas SDK

ApsaraMQ for RocketMQ menyediakan satu kelas SDK untuk Simple consumers: SimpleConsumer. Kelas ini tidak dapat mengonsumsi pesan dari topik Lite.

// Mengonsumsi pesan normal dengan SimpleConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        // Setel kelompok konsumen
        .setConsumerGroup("<your-consumer-group>")
        // Setel titik akses
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind langganan
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        .build();

try {
    // Tarik hingga 10 pesan, tunggu maksimal 30 detik
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        // Akui setiap pesan setelah pemrosesan berhasil
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // Tangani kegagalan seperti pembatasan kecepatan, lalu coba lagi pemanggilan receive
    e.printStackTrace();
}

Kontrol waktu pemrosesan untuk Push consumers

Pastikan pemrosesan pesan tetap dalam ambang batas timeout. Timeout yang sering menyebabkan retry tidak perlu dan pesan duplikat. Jika aplikasi Anda secara rutin menangani tugas berdurasi panjang, beralihlah ke Simple consumer dan atur durasi invisibility pesan yang sesuai.

Migrasi dari LitePullConsumer ke SimpleConsumer

LitePullConsumer adalah jenis konsumen dalam Apache RocketMQ 4.x Remoting SDK. Jenis ini telah dihapus pada SDK gRPC 5.x, dan SDK Go, Java, serta Python versi 5.x tidak menyediakan jenis ini.

Jika sebelumnya Anda menggunakan LitePullConsumer untuk konsumsi berbasis pull, migrasikan ke SimpleConsumer setelah melakukan upgrade ke SDK 5.x. SimpleConsumer menyediakan operasi berikut:

• ReceiveMessage: menarik pesan dari server sesuai permintaan.

• AckMessage: mengonfirmasi konsumsi yang berhasil.

• ChangeInvisibleDuration: mengubah durasi invisibility pesan untuk mengontrol interval retry.

Pada SDK Go, metode yang sesuai adalah Receive dan Ack. Pada SDK Java, metode yang sesuai adalah receive dan ack.