Implementasikan pemrosesan berurutan dengan pesan terurut - ApsaraMQ for RocketMQ

Pesan terurut adalah tipe pesan lanjutan di ApsaraMQ for RocketMQ. Topik ini menjelaskan skenario aplikasi, fitur, batasan, penggunaan, dan praktik terbaik untuk pesan terurut.

Skenario

Dalam skenario seperti pemrosesan event terurut, matchmaking dalam perdagangan, dan sinkronisasi data inkremental real-time, sistem heterogen harus mempertahankan konsistensi kuat melalui sinkronisasi status. Skenario ini mengharuskan perubahan event dari sistem hulu dikirim ke sistem hilir secara berurutan. Pesan terurut di ApsaraMQ for RocketMQ menjamin urutan transmisi data dalam kasus-kasus tersebut.

Skenario 1: Matchmaking dalam perdagangan

交易撮合

Misalnya, dalam perdagangan sekuritas dan saham, jika terdapat beberapa penawaran dengan harga yang sama, prinsip first-bid-first-trade berlaku. Sistem pemrosesan pesanan hilir harus memproses pesanan secara ketat sesuai urutan penawaran.

Skenario 2: Sinkronisasi data inkremental real-time

Misalnya, dalam skenario sinkronisasi perubahan database, database sumber hulu melakukan operasi tambah, hapus, dan ubah sesuai kebutuhan. Database tersebut mengirim log operasi biner sebagai pesan melalui ApsaraMQ for RocketMQ ke sistem pencarian hilir. Sistem hilir memulihkan data pesan secara berurutan, sehingga status data tetap diperbarui dalam urutan yang benar. Penggunaan pesan normal dapat menyebabkan status tidak terurut yang tidak sesuai dengan hasil operasi yang diharapkan. Pesan terurut menjamin bahwa status di sistem hilir konsisten dengan hasil operasi di sistem hulu.

Cara kerja

Apa itu pesan terurut?

Pesan terurut adalah tipe pesan lanjutan di ApsaraMQ for RocketMQ. Tipe pesan ini memungkinkan konsumen mengambil pesan dalam urutan yang sama dengan saat dikirim, sehingga mendukung pemrosesan sekuensial dalam skenario bisnis.

Berbeda dengan tipe pesan lainnya, pesan terurut menekankan hubungan sekuensial antara beberapa pesan selama pengiriman, penyimpanan, dan konsumsi.

ApsaraMQ for RocketMQ mengidentifikasi dan menentukan urutan pesan terurut menggunakan message group. Saat mengirim pesan terurut, Anda harus menetapkan message group untuk setiap pesan.

Penting

Urutan hanya dijamin untuk pesan dalam message group yang sama. Tidak ada hubungan sekuensial antara pesan dari message group berbeda atau pesan tanpa group.

Logika penentuan urutan berdasarkan message group memungkinkan partisi detail halus berdasarkan logika bisnis. Hal ini dapat meningkatkan paralelisme dan throughput sistem sekaligus memenuhi persyaratan sekuensial lokal.

Cara menjamin urutan pesan

Urutan pesan di ApsaraMQ for RocketMQ terdiri dari dua bagian: urutan produksi dan urutan konsumsi.

Urutan produksi: ApsaraMQ for RocketMQ menggunakan protokol antara produsen dan server untuk memastikan bahwa satu produsen mengirim pesan secara serial. Pesan tersebut kemudian disimpan dan dipersistensikan dalam urutan yang sama.
Untuk menjamin urutan produksi pesan, Anda harus memenuhi kondisi berikut:
- Message group yang sama
  
  Cakupan urutan produksi adalah message group. Produsen dapat menetapkan message group untuk setiap pesan yang dikirim. Urutan hanya dijamin untuk pesan dalam group yang sama. Pesan dari group berbeda atau tanpa group tidak dijamin urutannya.
- Produsen tunggal
  
  Urutan produksi hanya didukung untuk satu produsen. Jika produsen berbeda berada di sistem berbeda, urutan pesan mereka tidak dapat ditentukan, bahkan jika menggunakan message group yang sama.
- Pengiriman serial
  
  Klien produsen ApsaraMQ for RocketMQ mendukung akses multi-threaded. Namun, jika produsen menggunakan beberapa thread untuk mengirim pesan secara paralel, urutan pesan dari thread berbeda tidak dapat ditentukan.
Ketika produsen yang memenuhi kondisi ini mengirim pesan terurut ke ApsaraMQ for RocketMQ, pesan dengan message group yang sama disimpan dalam antrian yang sama sesuai urutan pengiriman. Logika penyimpanan di sisi server adalah sebagai berikut:
- Pesan dari message group yang sama disimpan dalam antrian yang sama secara berurutan.
- Pesan dari message group berbeda dapat dicampur dalam antrian yang sama, tetapi kontinuitasnya tidak dijamin.
Seperti yang ditunjukkan pada gambar, pesan dari Message Group 1 dan Message Group 4 dicampur dalam Queue 1. ApsaraMQ for RocketMQ menjamin bahwa pesan G1-M1, G1-M2, dan G1-M3 dari Message Group 1 disimpan sesuai urutan pengiriman. Demikian pula, pesan G4-M1 dan G4-M2 dari Message Group 4 disimpan secara berurutan. Namun, tidak ada hubungan sekuensial antara pesan dari Message Group 1 dan Message Group 4.
Urutan konsumsi: ApsaraMQ for RocketMQ menggunakan protokol antara konsumen dan server untuk memastikan bahwa pesan dikonsumsi secara ketat sesuai urutan penyimpanannya.
Untuk menjamin urutan konsumsi pesan, Anda harus memenuhi kondisi berikut:
- Pesanan Pengiriman
  
  ApsaraMQ for RocketMQ menggunakan kit pengembangan perangkat lunak (SDK) klien dan protokol komunikasi server untuk menjamin bahwa pesan dikirim sesuai urutan penyimpanannya. Namun, saat mengonsumsi pesan, bisnis Anda harus secara ketat mengikuti semantik receive-process-acknowledge untuk menghindari pesan tidak terurut akibat pemrosesan asinkron.
  
  Penting
  Ketika tipe konsumen adalah PushConsumer, ApsaraMQ for RocketMQ menjamin bahwa pesan dikirim ke konsumen satu per satu sesuai urutan penyimpanannya. Jika tipe konsumen adalah SimpleConsumer, konsumen mungkin menarik beberapa pesan sekaligus. Dalam kasus ini, bisnis Anda harus menjamin urutan konsumsi. Untuk informasi lebih lanjut tentang tipe konsumen, lihat Tipe konsumen.
- Batasan jumlah retry
  
  Untuk pesan terurut, ApsaraMQ for RocketMQ membatasi jumlah retry pengiriman. Jika pesan gagal dikonsumsi setelah mencapai jumlah maksimum retry, pesan tersebut dilewati. Hal ini mencegah pesan tersebut menghambat pesan berikutnya.
  
  Untuk skenario yang memerlukan urutan konsumsi ketat, tetapkan jumlah retry yang wajar untuk menghindari pesan tidak terurut akibat parameter yang tidak tepat.

Kombinasi urutan produksi dan konsumsi

Jika pesan harus diproses secara ketat dalam urutan masuk pertama, keluar pertama (FIFO), baik urutan produksi maupun konsumsi harus dijamin. Namun, dalam banyak skenario bisnis, satu produsen mungkin melayani beberapa konsumen hilir, dan tidak semuanya memerlukan konsumsi terurut. Anda dapat mengombinasikan urutan produksi dan konsumsi secara berbeda untuk berbagai skenario bisnis. Misalnya, Anda dapat mengirim pesan terurut tetapi menggunakan konsumsi konkuren non-terurut untuk meningkatkan throughput. Tabel berikut menjelaskan kombinasi lainnya:

Urutan produksi	Pesanan Konsumsi	Efek pengurutan
Tetapkan message group untuk menjamin pengiriman terurut.	Konsumsi terurut	Menjamin secara ketat urutan pesan di tingkat message group. Urutan konsumsi pesan dalam message group yang sama identik dengan urutan pengiriman.
Anda dapat menetapkan message group untuk menjamin pengiriman terurut.	Konsumsi konkuren	Pesan dikonsumsi secara konkuren dan diproses dalam urutan kronologis sejauh mungkin.
Tidak ada message group yang ditetapkan. Pesan dikirim tidak berurutan.	Konsumsi terurut	Pengurutan ketat di tingkat penyimpanan antrian. Berdasarkan properti antrian ApsaraMQ for RocketMQ, urutan konsumsi sesuai dengan urutan penyimpanan antrian tetapi belum tentu sesuai urutan pengiriman.
Tidak ada message group yang ditetapkan. Pesan dikirim tidak berurutan.	Konsumsi konkuren	Pesan dikonsumsi secara konkuren dan diproses dalam urutan kronologis berdasarkan upaya terbaik.

Siklus hidup pesan terurut

生命周期

Inisialisasi
Pesan dibuat dan diinisialisasi oleh produsen, siap dikirim ke broker.
Menunggu konsumsi
Pesan dikirim ke broker dan terlihat serta tersedia bagi konsumen.
Sedang dikonsumsi
Pesan diperoleh oleh konsumen dan diproses berdasarkan logika bisnis lokal konsumen.
Dalam proses ini, broker menunggu konsumen mengembalikan hasil konsumsi. Jika tidak ada respons dari konsumen dalam periode waktu tertentu, ApsaraMQ for RocketMQ melakukan retry terhadap pesan tersebut. Untuk informasi lebih lanjut, lihat Retry konsumsi.
Komitmen konsumsi
Konsumen menyelesaikan konsumsi dan mengirimkan hasil konsumsi ke broker. Broker menandai apakah pesan saat ini telah dikonsumsi.
Secara default, ApsaraMQ for RocketMQ menyimpan semua pesan. Saat hasil konsumsi dikirimkan, pesan ditandai sebagai telah dikonsumsi, bukan langsung dihapus. Pesan hanya dihapus jika periode retensi berakhir atau ruang penyimpanan sistem tidak mencukupi. Sebelum pesan dihapus, konsumen dapat mengonsumsi ulang pesan tersebut.
Penghapusan pesan
Jika periode retensi pesan berakhir atau ruang penyimpanan tidak mencukupi, ApsaraMQ for RocketMQ menghapus pesan yang paling awal disimpan dari file fisik secara bergulir. Untuk informasi lebih lanjut, lihat Penyimpanan dan pembersihan pesan.

Penting

Jika konsumsi pesan gagal atau timeout, logika retry server dipicu. Pesan yang diretry adalah pesan baru, dan siklus hidup pesan asli berakhir.
Ketika pesan terurut gagal dan diretry, pesan berikutnya tidak dapat dikonsumsi. Hal ini menjamin urutan pesan. Pesan tersebut hanya dapat diproses setelah pesan saat ini berhasil dikonsumsi.

Batasan

Pesan terurut hanya dapat dikirim ke topik yang memiliki MessageType diatur ke FIFO. Tipe pesan harus sesuai dengan tipe topik.

Penting

Jika kelompok konsumen diatur untuk menggunakan pengiriman terurut, semua pesan yang dikonsumsi oleh kelompok tersebut akan ditagih sebagai pesan terurut, terlepas dari tipe pesannya. Jika bisnis Anda tidak memerlukan konsumsi terurut secara ketat, Anda dapat mengatur mode pengiriman kelompok ke pengiriman konkuren untuk mengurangi biaya.

Optimasi konkurensi konsumsi

Anda harus menggunakan RocketMQ 5.x gRPC SDK. Saat melakukan konsumsi terurut, klien PushConsumer dalam rangkaian SDK ini dapat menetapkan pesan dari MessageQueue yang sama ke thread berbeda untuk konsumsi konkuren berdasarkan MessageGroup-nya. Hal ini sangat meningkatkan konkurensi konsumsi untuk pesan terurut. Semakin tersebar nilai MessageGroup, semakin besar peningkatannya. Versi SDK yang didukung adalah sebagai berikut:

Java SDK: 5.0.8 dan versi lebih baru.
C++ SDK: 5.0.3 dan versi lebih baru.
SDK lainnya: Tidak didukung.

Contoh penggunaan

Berbeda dengan pesan normal, pesan terurut mengharuskan Anda menentukan message group saat mengirimnya. Rancang message group dengan granularitas se-detail mungkin berdasarkan skenario bisnis Anda untuk mendukung partisi bisnis yang efektif dan skalabilitas konkuren.

Kode Java berikut menunjukkan cara mengirim dan menerima pesan terurut:

Untuk kode contoh lengkap pengiriman dan penerimaan pesan, lihat RocketMQ 5.x gRPC SDK.

Kode contoh

Kirim pesan terurut

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;


public class ProducerExample {
    public static void main(String[] args) throws ClientException {
        /**
         * Titik akhir instans. Dapatkan dari tab Endpoints pada halaman detail instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, gunakan titik akhir VPC.
         * Jika Anda mengakses instans dari mesin lokal melalui jaringan publik atau dari pusat data lokal, gunakan titik akhir publik. Untuk menggunakan titik akhir publik, Anda harus mengaktifkan akses jaringan publik untuk instans tersebut.
         */
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // Nama topik tempat pesan dikirim. Anda harus membuat topik di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan topik yang belum dibuat.
        String topic = "Your Topic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda mengakses instans melalui jaringan publik, Anda juga harus menyetel username dan password instans dalam konfigurasi. Dapatkan username dan password dari tab Intelligent Authentication pada halaman Access Control di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, Anda tidak perlu mengonfigurasi ini. Server secara otomatis mendapatkan informasi berdasarkan VPC.
         * Jika Anda menggunakan instans Serverless, Anda harus menyetel username dan password untuk akses jaringan publik. Jika akses bebas autentikasi melalui jaringan internal diaktifkan, Anda tidak perlu menyetel username dan password untuk akses jaringan internal.
         */
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        ClientConfiguration configuration = builder.build();
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();
         // Kirim pesan terurut.
        Message message = provider.newMessageBuilder()
                .setTopic("topic")
                // Tetapkan kunci pesan. Anda dapat menggunakan kunci ini untuk menemukan pesan secara akurat.
                .setKeys("messageKey")
                // Tetapkan tag pesan. Konsumen dapat menggunakan tag ini untuk memfilter pesan.
                .setTag("messageTag")
                // Tetapkan message group untuk pesan terurut. Buat group se-discrete mungkin untuk menghindari hot spot groups.
                .setMessageGroup("fifoGroup001")
                // Isi pesan.
                .setBody("messageBody".getBytes())
                .build();
        try {
            // Kirim pesan. Perhatikan hasil pengiriman dan tangkap exception seperti kegagalan.
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

Konsumsi dengan PushConsumer

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        /**
         * Titik akhir instans. Dapatkan dari tab Endpoints pada halaman detail instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, gunakan titik akhir VPC.
         * Jika Anda mengakses instans dari mesin lokal melalui jaringan publik atau dari pusat data lokal, gunakan titik akhir publik. Untuk menggunakan titik akhir publik, Anda harus mengaktifkan akses jaringan publik untuk instans tersebut.
         */
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // Topik yang akan di-subscribe. Anda harus membuat topik di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan topik yang belum dibuat.
        String topic = "Your Topic";
        // Kelompok konsumen untuk konsumen. Anda harus membuat kelompok di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan kelompok yang belum dibuat.
        String consumerGroup = "Your ConsumerGroup";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda mengakses instans melalui jaringan publik, Anda juga harus menyetel username dan password instans dalam konfigurasi. Dapatkan username dan password dari tab Intelligent Authentication pada halaman Access Control di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, Anda tidak perlu mengonfigurasi ini. Server secara otomatis mendapatkan informasi berdasarkan VPC.
         * Jika Anda menggunakan instans Serverless, Anda harus menyetel username dan password untuk akses jaringan publik. Jika akses bebas autentikasi melalui jaringan internal diaktifkan, Anda tidak perlu menyetel username dan password untuk akses jaringan internal.
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
        ClientConfiguration clientConfiguration = builder.build();
        // Aturan filter untuk subscription pesan. Ini berarti semua tag di-subscribe.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Inisialisasi PushConsumer. Anda perlu mengikat kelompok konsumen, parameter komunikasi, dan hubungan subscription.
        // Saat mengonsumsi pesan terurut, pastikan kelompok konsumen saat ini berada dalam mode pengiriman terurut. Jika tidak, pesan tetap dikirim secara konkuren dan tidak terurut.
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Tetapkan kelompok konsumen.
                .setConsumerGroup(consumerGroup)
                // Tetapkan hubungan subscription yang telah ditentukan sebelumnya.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Tetapkan listener konsumsi.
                .setMessageListener(messageView -> {
                    // Proses pesan dan kembalikan hasil konsumsi.
                    // LOGGER.info("Consume message={}", messageView);
                    System.out.println("Consume Message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();
        Thread.sleep(Long.MAX_VALUE);
        // Jika PushConsumer tidak lagi diperlukan, tutup prosesnya.
        //pushConsumer.close();
    }
}

Konsumsi dengan SimpleConsumer

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    private SimpleConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException {
        /**
         * Titik akhir instans. Dapatkan dari tab Endpoints pada halaman detail instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, gunakan titik akhir VPC.
         * Jika Anda mengakses instans dari mesin lokal melalui jaringan publik atau dari pusat data lokal, gunakan titik akhir publik. Untuk menggunakan titik akhir publik, Anda harus mengaktifkan akses jaringan publik untuk instans tersebut.
         */
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // Topik yang akan di-subscribe. Anda harus membuat topik di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan topik yang belum dibuat.
        String topic = "Your Topic";
        // Kelompok konsumen untuk konsumen. Anda harus membuat kelompok di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan kelompok yang belum dibuat.
        String consumerGroup = "Your ConsumerGroup";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda mengakses instans melalui jaringan publik, Anda juga harus menyetel username dan password instans dalam konfigurasi. Dapatkan username dan password dari tab Intelligent Authentication pada halaman Access Control di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, Anda tidak perlu mengonfigurasi ini. Server secara otomatis mendapatkan informasi berdasarkan VPC.
         * Jika Anda menggunakan instans Serverless, Anda harus menyetel username dan password untuk akses jaringan publik. Jika akses bebas autentikasi melalui jaringan internal diaktifkan, Anda tidak perlu menyetel username dan password untuk akses jaringan internal.
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
        ClientConfiguration clientConfiguration = builder.build();

        Duration awaitDuration = Duration.ofSeconds(10);
        // Aturan filter untuk subscription pesan. Ini berarti semua tag di-subscribe.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Inisialisasi SimpleConsumer. Anda perlu mengikat kelompok konsumen, parameter komunikasi, dan hubungan subscription.
        // Saat mengonsumsi pesan terurut, pastikan kelompok konsumen saat ini berada dalam mode pengiriman terurut. Jika tidak, pesan tetap dikirim secara konkuren dan tidak terurut.
        SimpleConsumer consumer = provider.newSimpleConsumerBuilder().setClientConfiguration(clientConfiguration)
                // Tetapkan kelompok konsumen.
                .setConsumerGroup(consumerGroup)
                // Tetapkan timeout long polling.
                .setAwaitDuration(awaitDuration)
                // Tetapkan hubungan subscription yang telah ditentukan sebelumnya.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)).build();
        // Tetapkan jumlah maksimum pesan yang akan ditarik kali ini.
        int maxMessageNum = 16;
        // Tetapkan durasi invisibilitas untuk pesan.
        Duration invisibleDuration = Duration.ofSeconds(10);
        // SimpleConsumer mengharuskan klien secara aktif melakukan loop untuk mendapatkan dan memproses pesan.
        // Untuk meningkatkan konsumsi real-time, gunakan beberapa thread untuk menarik pesan secara konkuren.
        while (true) {
            // Perhatikan bahwa untuk pesan dalam MessageGroup yang sama, jika pesan sebelumnya belum dikonsumsi, Anda tidak dapat mendapatkan pesan berikutnya dengan memanggil receive lagi.
            final List<MessageView> messageViewList = consumer.receive(maxMessageNum, invisibleDuration);
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // Setelah pemrosesan selesai, Anda harus secara aktif memanggil ack untuk mengirimkan hasil konsumsi.
                try {
                    consumer.ack(messageView);
                } catch (ClientException e) {
                    // Jika penarikan gagal karena pembatasan kecepatan sistem atau alasan lain, Anda perlu menginisiasi ulang permintaan untuk mendapatkan pesan.
                    e.printStackTrace();
                }
            });
        }
        // Jika SimpleConsumer tidak lagi diperlukan, tutup prosesnya.
        // consumer.close();
    }
}

Dapatkan log untuk retry konsumsi pesan terurut

Retry untuk konsumsi terurut oleh PushConsumer terjadi di klien konsumen. Server tidak dapat mengambil log detail untuk retry konsumsi. Jika hasil pengiriman pesan terurut dalam jejak pesan adalah "failed", periksa log klien konsumen untuk informasi seperti jumlah maksimum retry dan klien konsumen.

Untuk informasi lebih lanjut tentang path log klien konsumen, lihat Konfigurasi log.

Anda dapat mencari kata kunci berikut dalam log klien untuk dengan cepat menemukan konten terkait kegagalan konsumsi:

Message listener raised an exception while consuming messages
Failed to consume fifo message finally, run out of attempt times

Praktik terbaik

Gunakan konsumsi serial dan hindari konsumsi batch untuk mencegah pemrosesan tidak terurut

Konsumsi pesan secara serial. Hindari mengonsumsi beberapa pesan sekaligus, karena hal ini dapat menyebabkan pemrosesan tidak terurut.

Misalnya, pesan dikirim dalam urutan 1->2->3->4. Selama konsumsi batch, urutannya menjadi 1->[2,3](pemrosesan batch gagal)->[2,3](retry)->4. Dalam skenario ini, kegagalan pesan 3 dapat menyebabkan pesan 2 diproses lagi, menghasilkan konsumsi tidak terurut.

Distribusikan message group untuk menghindari hot spot

ApsaraMQ for RocketMQ menjamin bahwa pesan dalam message group yang sama disimpan dalam antrian yang sama. Jika pesan dari berbagai skenario bisnis terkonsentrasi dalam satu atau beberapa message group, tekanan penyimpanan akan terfokus pada antrian server tertentu. Hal ini dapat dengan mudah menciptakan hot spot performa dan menghambat skalabilitas. Rekomendasi umum untuk merancang message group adalah menggunakan ID pesanan atau ID pengguna sebagai dasar pengurutan. Hal ini menjamin bahwa pesan untuk end user yang sama diproses secara berurutan, sedangkan pesan untuk pengguna berbeda tidak memerlukan pengurutan.

Oleh karena itu, partisi bisnis Anda pada granularitas message group. Misalnya, gunakan ID pesanan atau ID pengguna sebagai kunci message group. Hal ini memungkinkan pesan untuk end user yang sama diproses secara berurutan, sedangkan pesan untuk pengguna berbeda tidak memerlukan pengurutan.