ApsaraMQ for RocketMQ：Jenis konsumen

Penggunaan

Menggunakan Push consumer cukup mudah. Anda mendaftarkan listener pesan selama inisialisasi konsumen dan mengimplementasikan logika pemrosesan pesan di dalam listener tersebut. SDK ApsaraMQ for RocketMQ menangani pengambilan pesan, pemanggilan listener, dan pemrosesan ulang di latar belakang. Untuk contoh kode, lihat kelas SDK yang sesuai.

Mekanisme internal

Untuk Push consumer, pemrosesan pesan real-time didasarkan pada model thread Reactor khas dalam SDK. Seperti yang ditunjukkan pada gambar berikut, SDK memiliki thread long-polling bawaan. Thread ini menarik pesan secara asinkron ke dalam antrian cache internal SDK. Pesan-pesan tersebut kemudian dikirimkan ke thread konsumen, yang memicu listener untuk menjalankan logika konsumsi lokal.

Keandalan dan retry

Untuk Push consumer, satu-satunya batas antara SDK klien dan logika konsumsi adalah antarmuka listener pesan. SDK klien secara ketat menentukan apakah sebuah pesan berhasil dikonsumsi berdasarkan hasil yang dikembalikan oleh listener dan melakukan retry untuk memastikan keandalan. Semua pesan harus diproses secara sinkron. Hasil panggilan harus dikembalikan ketika antarmuka listener selesai. Distribusi asinkron tidak diperbolehkan. Untuk informasi lebih lanjut tentang retry pesan, lihat Kebijakan retry PushConsumer.

Jaminan urutan

Berdasarkan definisi pesan terurut di ApsaraMQ for RocketMQ, jika kelompok konsumen diatur ke mode konsumsi terurut, Push consumer secara ketat mengikuti urutan pesan saat memanggil listener pesan. Urutan konsumsi dijamin tanpa perubahan apa pun pada logika bisnis.

Skenario

Kelas SDK yang sesuai

ApsaraMQ for RocketMQ menyediakan dua kelas SDK untuk Push consumer: PushConsumer dan LitePushConsumer.

• PushConsumer: Cocok untuk mengonsumsi pesan dari topik yang bukan tipe Lite.

• LitePushConsumer: Khusus untuk mengonsumsi pesan dari topik tipe Lite. Memungkinkan kontrol konsumsi pada tingkat granularitas topik Lite.

// Contoh konsumsi: Gunakan PushConsumer untuk mengonsumsi pesan normal.
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 ConsumerGroup")
                // Setel titik akhir.
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                // Setel hubungan langganan yang telah ditentukan sebelumnya.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Setel listener pesan.
                .setMessageListener(new MessageListener() {
                    @Override
                    public ConsumeResult consume(MessageView messageView) {
                        // Konsumsi pesan dan kembalikan hasil pemrosesan.
                        return ConsumeResult.SUCCESS;
                    }
                })
                .build();

// Contoh konsumsi: Gunakan LitePushConsumer untuk mengonsumsi pesan normal.
ClientServiceProvider provider = ClientServiceProvider.loadService();

 LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
        // Setel titik akhir.
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
        // Setel topik.
        .bindTopic("Your Topic")
        // Setel kelompok konsumen.
        .setConsumerGroup("Your ConsumerGroup")
        // Setel listener pesan.
        .setMessageListener(messageView -> {
            // Konsumsi pesan dan kembalikan hasil pemrosesan.
            return ConsumeResult.SUCCESS;
        })
        .build();

// Berlangganan ke topik Lite target.
litePushConsumer.subscribeLite("Your Lite Topic 1");
litePushConsumer.subscribeLite("Your Lite Topic 2");

Penggunaan

Menggunakan Simple consumer melibatkan pemanggilan beberapa operasi API dari logika bisnis Anda. Anda memanggil operasi untuk mengambil pesan dan mendistribusikannya ke thread bisnis untuk diproses. Setelah pemrosesan selesai, Anda memanggil operasi commit untuk mengembalikan hasil ke server. Untuk contoh kode, lihat kelas SDK yang sesuai.

Retry untuk keandalan

Untuk Simple consumer, kit pengembangan perangkat lunak (SDK) klien dan server berkomunikasi menggunakan operasi ReceiveMessage dan AckMessage. Jika SDK klien berhasil memproses pesan, SDK tersebut memanggil operasi AckMessage. Jika pemrosesan gagal, jangan mengirim respons ACK. Hal ini akan memicu retry setelah durasi invisibilitas pesan berakhir. Untuk informasi lebih lanjut, lihat Kebijakan retry SimpleConsumer.

Urutan pesan yang dijamin

Simple consumer memproses pesan terurut dari ApsaraMQ for RocketMQ sesuai urutan penyimpanannya. Untuk sekelompok pesan yang harus tetap terurut, jika pesan sebelumnya belum diproses, pesan berikutnya tidak dapat diambil.

Skenario

SimpleConsumer menyediakan operasi API atomik untuk mengambil pesan dan mengirimkan hasil konsumsi. Metode ini lebih fleksibel dibandingkan menggunakan PushConsumer. SimpleConsumer cocok untuk skenario berikut:

• Durasi pemrosesan pesan tidak dapat diprediksi: Gunakan SimpleConsumer jika durasi pemrosesan pesan tidak dapat diperkirakan atau jika pesan sering memerlukan waktu lama untuk diproses. Anda dapat menentukan perkiraan durasi pemrosesan selama konsumsi. Jika perkiraan tersebut tidak sesuai dengan kebutuhan bisnis Anda, Anda dapat mengubahnya dengan memanggil operasi API.

• Skenario kustom tingkat lanjut: SDK SimpleConsumer tidak memiliki enkapsulasi thread yang kompleks. Logika bisnis Anda memiliki kontrol penuh. Hal ini memungkinkan Anda mengimplementasikan skenario tingkat lanjut seperti distribusi asinkron dan konsumsi batch.

• Laju konsumsi kustom: Dengan SimpleConsumer, logika bisnis Anda secara aktif memanggil operasi untuk mengambil pesan. Hal ini memungkinkan Anda menyesuaikan frekuensi pengambilan pesan untuk mengontrol laju konsumsi.

Kelas SDK yang sesuai

ApsaraMQ for RocketMQ menyediakan satu kelas SDK untuk Simple consumer: SimpleConsumer. Kelas ini menawarkan berbagai operasi kustom. SimpleConsumer tidak dapat mengonsumsi pesan dari topik Lite.

// Contoh konsumsi: Gunakan SimpleConsumer untuk mengonsumsi pesan normal. Secara aktif ambil pesan, proses, dan kirimkan hasilnya.
        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 ConsumerGroup")
                // Setel titik akhir.
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                // Setel hubungan langganan yang telah ditentukan sebelumnya.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        List<MessageView> messageViewList = null;
        try {
            // SimpleConsumer harus secara aktif mengambil dan memproses pesan.
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // Setelah pemrosesan selesai, aktif panggil ack() untuk mengirimkan hasil konsumsi.
                try {
                    simpleConsumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            // Jika pengambilan gagal karena alasan seperti throttling sistem, Anda harus menginisiasi ulang permintaan untuk mengambil pesan.
            e.printStackTrace();
        }

Kontrol waktu konsumsi untuk push consumer

Kontrol secara ketat waktu konsumsi pesan untuk Push consumer guna menghindari pemrosesan ulang akibat timeout. Jika aplikasi Anda sering memproses pesan yang memerlukan waktu lama, gunakan Simple consumer dan atur durasi invisibilitas pesan.