arsitektur pesan terurut - ApsaraMQ for RocketMQ

Pesan terurut merupakan fitur lanjutan dalam ApsaraMQ for RocketMQ. Topik ini menjelaskan skenario, prinsip kerja, batasan, penggunaan, dan praktik terbaik untuk pesan terurut.

Skenario

Dalam skenario seperti pemrosesan event berurutan, pencocokan transaksi, dan sinkronisasi data inkremental secara real-time, sistem heterogen harus mempertahankan konsistensi kuat melalui sinkronisasi status. Dalam skenario ini, perubahan event di sistem hulu harus dikirimkan ke sistem hilir untuk diproses secara berurutan. Anda dapat menggunakan pesan terurut di ApsaraMQ for RocketMQ untuk memastikan transmisi data berurutan.

Skenario 1: Pencocokan transaksi

交易撮合

Dalam skenario pencocokan transaksi sekuritas dan saham, jika terdapat beberapa penawaran dengan harga yang sama, prinsip first-in, first-out (FIFO) berlaku. Sistem pemrosesan pesanan di hilir harus memproses pesanan secara ketat sesuai urutan penawaran diajukan.

Skenario 2: Sinkronisasi data inkremental real-time

Dalam skenario sinkronisasi inkremental perubahan database, database sumber di hulu melakukan operasi tambah, hapus, dan ubah. Log operasi biner dikirim sebagai pesan melalui ApsaraMQ for RocketMQ ke sistem pencarian di hilir. Sistem hilir kemudian memulihkan data secara berurutan dari pesan-pesan tersebut untuk memperbarui statusnya. Jika Anda menggunakan pesan normal, status tersebut mungkin menjadi tidak konsisten dengan hasil operasi yang diharapkan. Pesan terurut memastikan bahwa status di hilir konsisten dengan hasil operasi di hulu.

Cara kerja

Apa itu pesan terurut?

Pesan terurut adalah jenis pesan lanjutan dalam ApsaraMQ for RocketMQ. Pesan ini memungkinkan konsumen menerima pesan dalam urutan yang sama saat dikirim, sehingga mendukung pemrosesan berurutan dalam skenario bisnis Anda.

Berbeda dengan jenis pesan lainnya, pesan terurut menjamin hubungan berurutan antara beberapa pesan selama pengiriman, penyimpanan, dan pengiriman.

Dalam ApsaraMQ for RocketMQ, urutan pesan ditentukan oleh kelompok pesan. Saat mengirim pesan terurut, Anda harus menentukan kelompok pesan untuk setiap pesan.

Penting

Urutan hanya dijamin untuk pesan dalam kelompok pesan yang sama. Urutan pesan dari kelompok pesan berbeda atau pesan tanpa kelompok yang ditentukan tidak dijamin.

Logika pengurutan berbasis kelompok pesan memungkinkan pemisahan detail halus berdasarkan logika bisnis Anda. Hal ini dapat meningkatkan paralelisme dan throughput sistem sekaligus memastikan urutan bisnis lokal.

Cara menjamin urutan pesan

Pengurutan pesan dalam ApsaraMQ for RocketMQ melibatkan dua tahap: urutan produksi dan urutan konsumsi.

Urutan produksi: ApsaraMQ for RocketMQ menggunakan protokol antara produsen dan server untuk memastikan bahwa pesan yang dikirim secara serial oleh satu produsen disimpan dan dipertahankan dalam urutan tersebut.
Untuk menjamin urutan produksi pesan, Anda harus memenuhi kondisi berikut:
- Kelompok pesan yang sama
  Urutan produksi hanya berlaku untuk satu kelompok pesan. Saat produsen mengirim pesan, produsen dapat menetapkan kelompok pesan untuk setiap pesan. Urutan hanya dijamin untuk pesan dalam kelompok pesan yang sama. Urutan pesan dari kelompok pesan berbeda atau pesan tanpa kelompok yang ditentukan tidak dijamin.
- Produsen tunggal
  Urutan produksi hanya dijamin untuk satu produsen. Jika produsen berbeda mengirim pesan ke kelompok pesan yang sama, bahkan jika tersebar di sistem berbeda, urutan pesan tidak dapat dijamin.
- Pengiriman serial
  Klien produsen ApsaraMQ for RocketMQ mendukung akses aman multi-threaded. Namun, jika produsen menggunakan banyak thread untuk mengirim pesan secara konkuren, urutan pesan yang dikirim oleh thread berbeda tidak dapat dijamin.
Saat produsen yang memenuhi kondisi ini mengirim pesan terurut ke ApsaraMQ for RocketMQ, pesan dengan kelompok pesan yang sama disimpan dalam antrian yang sama sesuai urutan pengiriman. Logika penyimpanan berurutan server adalah sebagai berikut:
- Pesan dari kelompok pesan yang sama disimpan dalam antrian yang sama secara berurutan.
- Pesan dari kelompok pesan berbeda dapat dicampur dalam antrian yang sama, tetapi kontinuitasnya tidak dijamin.
Seperti yang ditunjukkan pada gambar sebelumnya, pesan dari Kelompok Pesan 1 dan Kelompok Pesan 4 dicampur dalam Antrian 1. ApsaraMQ for RocketMQ memastikan bahwa pesan G1-M1, G1-M2, dan G1-M3 dari Kelompok Pesan 1 disimpan sesuai urutan pengiriman. Pesan G4-M1 dan G4-M2 dari Kelompok Pesan 4 juga disimpan berurutan. Namun, urutan antara pesan dari Kelompok Pesan 1 dan Kelompok Pesan 4 tidak dijamin.
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:
- Urutan pengiriman
  ApsaraMQ for RocketMQ menggunakan SDK klien dan protokol komunikasi server untuk memastikan bahwa pesan dikirimkan sesuai urutan penyimpanannya. Namun, saat aplikasi Anda mengonsumsi pesan, aplikasi harus secara ketat mengikuti semantik terima-proses-konfirmasi untuk mencegah gangguan urutan akibat pemrosesan asinkron.
  Penting
  Jika tipe konsumen adalah PushConsumer, ApsaraMQ for RocketMQ memastikan bahwa pesan dikirimkan ke konsumen satu per satu sesuai urutan penyimpanannya. Jika tipe konsumen adalah SimpleConsumer, konsumen mungkin menarik beberapa pesan sekaligus. Dalam kasus ini, aplikasi Anda harus memastikan urutan konsumsi. Untuk informasi lebih lanjut tentang tipe konsumen, lihat Tipe konsumen.
- Percobaan ulang terbatas
  Pengiriman pesan terurut dalam ApsaraMQ for RocketMQ dibatasi pada jumlah retrai tertentu. Jika sebuah pesan terus-menerus gagal dikonsumsi dan melebihi jumlah maksimum retrai, pesan tersebut tidak akan diretrai lagi. Sistem melewati pesan ini untuk mencegahnya menghambat pemrosesan pesan berikutnya.
  Untuk skenario yang memerlukan urutan konsumsi ketat, Anda dapat menetapkan jumlah retrai yang wajar untuk mencegah gangguan urutan akibat pengaturan parameter yang salah.

Kombinasi urutan produksi dan konsumsi

Jika pesan harus diproses secara ketat sesuai prinsip FIFO, baik urutan produksi maupun konsumsi harus dijamin. Namun, dalam skenario bisnis umum, satu produsen mungkin terhubung ke banyak konsumen di hilir, yang tidak semuanya memerlukan konsumsi berurutan. Anda dapat menggunakan kombinasi berbeda antara urutan produksi dan konsumsi untuk berbagai skenario bisnis. Misalnya, Anda dapat mengirim pesan terurut tetapi menggunakan konsumsi konkuren untuk meningkatkan throughput. Tabel berikut menjelaskan beberapa kombinasi:

Urutan produksi	Pesanan Pemakaian	Efek pengurutan
Pengiriman terurut (kelompok pesan ditetapkan).	Konsumsi terurut	Urutan pesan ketat dijamin pada tingkat kelompok pesan. Urutan konsumsi identik dengan urutan pengiriman untuk pesan dalam kelompok pesan yang sama.
Pengiriman terurut (kelompok pesan ditetapkan).	Konsumsi konkuren	Konsumsi konkuren. Pesan diproses sedekat mungkin secara kronologis, tetapi urutan tidak dijamin.
Pengiriman tidak terurut (tidak ada kelompok pesan yang ditetapkan).	Konsumsi terurut	Urutan ketat pada tingkat antrian. Berdasarkan properti antrian ApsaraMQ for RocketMQ, urutan konsumsi sesuai dengan urutan penyimpanan dalam antrian, tetapi tidak dijamin sesuai dengan urutan pengiriman.
Pengiriman tidak terurut (tidak ada kelompok pesan yang ditetapkan).	Konsumsi konkuren	Konsumsi konkuren. Pesan diproses sedekat mungkin secara kronologis, tetapi urutan tidak dijamin.

Siklus hidup pesan terurut

生命周期

Inisialisasi
Pesan dibuat dan diinisialisasi oleh produsen dan siap dikirim ke broker.
Tertunda 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 tanggapan dari konsumen dalam periode waktu tertentu, ApsaraMQ for RocketMQ melakukan retrai terhadap pesan tersebut. Untuk informasi lebih lanjut, lihat Retrai 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 alih-alih 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

Saat pesan gagal dikonsumsi atau waktu konsumsi habis, logika retrai server dipicu. Pesan yang diretrai dianggap sebagai pesan baru, dan siklus hidup pesan asli berakhir.
Untuk menjamin urutan pesan, saat pesan terurut yang gagal diretrai, pesan berikutnya dalam kelompok pesan yang sama diblokir. Pesan tersebut hanya dapat dikonsumsi setelah pesan saat ini berhasil dikonsumsi.

Batasan

Anda hanya dapat mengirim pesan terurut ke topik yang MessageType-nya diatur ke FIFO. Jenis pesan harus sesuai dengan jenis topik.

Optimasi konkurensi konsumsi

Klien Anda harus menggunakan RocketMQ 5.x gRPC SDK. Saat mengonsumsi pesan terurut, klien PushConsumer dalam rangkaian SDK ini dapat menugaskan pesan dari MessageQueue yang sama ke thread berbeda untuk konsumsi konkuren berdasarkan MessageGroup-nya. Fitur ini secara signifikan meningkatkan konkurensi konsumsi pesan terurut. Peningkatan lebih signifikan ketika nilai MessageGroup lebih diskrit. Versi SDK yang didukung adalah sebagai berikut:

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

Contoh

Berbeda dengan pengiriman pesan normal, Anda harus menetapkan kelompok pesan saat mengirim pesan terurut. Anda dapat merancang kelompok pesan dengan granularitas sehalus mungkin berdasarkan skenario bisnis Anda. Hal ini memfasilitasi pemisahan logika bisnis dan meningkatkan skalabilitas konkuren.

Kode contoh berikut menunjukkan cara mengirim dan menerima pesan terurut dalam Java:

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. Anda dapat memperoleh titik akhir dari tab Titik Akhir pada halaman Detail Instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, kami menyarankan agar Anda menggunakan titik akhir VPC.
         * Jika Anda mengakses instans dari komputer lokal atau pusat data melalui internet, Anda dapat menggunakan 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 tujuan untuk pesan. Anda harus membuat topik di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan topik yang tidak ada.
        String topic = "Your Topic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda menggunakan titik akhir publik, Anda juga harus mengonfigurasi nama pengguna dan kata sandi untuk instans. Anda dapat memperoleh nama pengguna dan kata sandi dari tab Otentikasi Cerdas pada halaman Kontrol Akses untuk instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, Anda tidak perlu mengonfigurasi ini. Server secara otomatis memperoleh informasi berdasarkan VPC.
         * Jika instans adalah instans arsitektur tanpa server, Anda harus menetapkan nama pengguna dan kata sandi untuk akses jaringan publik. Jika akses bebas otentikasi melalui jaringan internal diaktifkan, Anda tidak perlu menetapkan nama pengguna dan kata sandi 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 tertentu secara akurat.
                .setKeys("messageKey")
                // Tetapkan tag pesan. Konsumen dapat menggunakan tag ini untuk memfilter pesan.
                .setTag("messageTag")
                // Tetapkan kelompok pesan untuk pesan terurut. Usahakan kelompok ini se-diskrit mungkin untuk menghindari kelompok pesan hot spot.
                .setMessageGroup("fifoGroup001")
                // Isi pesan.
                .setBody("messageBody".getBytes())
                .build();
        try {
            // Kirim pesan. Perhatikan hasil pengiriman dan tangkap pengecualian seperti kegagalan.
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

Konsumsi pesan 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. Anda dapat memperoleh titik akhir dari tab Titik Akhir pada halaman Detail Instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, kami menyarankan agar Anda menggunakan titik akhir VPC.
         * Jika Anda mengakses instans dari komputer lokal atau pusat data melalui internet, Anda dapat menggunakan 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 tujuan yang ingin Anda langganan. Anda harus membuat topik di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan topik yang tidak ada.
        String topic = "Your Topic";
        // Kelompok konsumen tempat konsumen berada. Anda harus membuat kelompok di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan kelompok yang tidak ada.
        String consumerGroup = "Your ConsumerGroup";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda menggunakan titik akhir publik, Anda juga harus mengonfigurasi nama pengguna dan kata sandi untuk instans. Anda dapat memperoleh nama pengguna dan kata sandi dari tab Otentikasi Cerdas pada halaman Kontrol Akses untuk instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, Anda tidak perlu mengonfigurasi ini. Server secara otomatis memperoleh informasi berdasarkan VPC.
         * Jika instans adalah instans arsitektur tanpa server, Anda harus menetapkan nama pengguna dan kata sandi untuk akses jaringan publik. Jika akses bebas otentikasi melalui jaringan internal diaktifkan, Anda tidak perlu menetapkan nama pengguna dan kata sandi untuk akses jaringan internal.
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
        ClientConfiguration clientConfiguration = builder.build();
        // Ekspresi filter untuk langganan pesan. Ini menunjukkan bahwa semua pesan dengan tag apa pun dilangganan.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Inisialisasi PushConsumer. Anda harus menyambungkan kelompok konsumen, parameter komunikasi, dan hubungan langganan.
        // Saat Anda mengonsumsi pesan terurut, pastikan kelompok konsumen saat ini berada dalam mode pengiriman terurut. Jika tidak, pesan tetap dikirimkan secara konkuren dan tidak berurutan.
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Tetapkan kelompok konsumen.
                .setConsumerGroup(consumerGroup)
                // Tetapkan hubungan langganan yang telah ditautkan sebelumnya.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Tetapkan pendengar pesan.
                .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 Anda tidak lagi memerlukan PushConsumer, Anda dapat mematikan proses.
        //pushConsumer.close();
    }
}

Konsumsi pesan 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. Anda dapat memperoleh titik akhir dari tab Titik Akhir pada halaman Detail Instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, kami menyarankan agar Anda menggunakan titik akhir VPC.
         * Jika Anda mengakses instans dari komputer lokal atau pusat data melalui internet, Anda dapat menggunakan 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 tujuan yang ingin Anda langganan. Anda harus membuat topik di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan topik yang tidak ada.
        String topic = "Your Topic";
        // Kelompok konsumen tempat konsumen berada. Anda harus membuat kelompok di Konsol terlebih dahulu. Kesalahan akan dikembalikan jika Anda menggunakan kelompok yang tidak ada.
        String consumerGroup = "Your ConsumerGroup";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * Jika Anda menggunakan titik akhir publik, Anda juga harus mengonfigurasi nama pengguna dan kata sandi untuk instans. Anda dapat memperoleh nama pengguna dan kata sandi dari tab Otentikasi Cerdas pada halaman Kontrol Akses untuk instans di Konsol.
         * Jika Anda mengakses instans dari Instance ECS Alibaba Cloud melalui jaringan internal, Anda tidak perlu mengonfigurasi ini. Server secara otomatis memperoleh informasi berdasarkan VPC.
         * Jika instans adalah instans arsitektur tanpa server, Anda harus menetapkan nama pengguna dan kata sandi untuk akses jaringan publik. Jika akses bebas otentikasi melalui jaringan internal diaktifkan, Anda tidak perlu menetapkan nama pengguna dan kata sandi untuk akses jaringan internal.
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
        ClientConfiguration clientConfiguration = builder.build();

        Duration awaitDuration = Duration.ofSeconds(10);
        // Ekspresi filter untuk langganan pesan. Ini menunjukkan bahwa semua pesan dengan tag apa pun dilangganan.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Inisialisasi SimpleConsumer. Anda harus menyambungkan kelompok konsumen, parameter komunikasi, dan hubungan langganan.
        // Saat Anda mengonsumsi pesan terurut, pastikan kelompok konsumen saat ini berada dalam mode pengiriman terurut. Jika tidak, pesan tetap dikirimkan secara konkuren dan tidak berurutan.
        SimpleConsumer consumer = provider.newSimpleConsumerBuilder().setClientConfiguration(clientConfiguration)
                // Tetapkan kelompok konsumen.
                .setConsumerGroup(consumerGroup)
                // Tetapkan periode timeout polling panjang.
                .setAwaitDuration(awaitDuration)
                // Tetapkan hubungan langganan yang telah ditautkan sebelumnya.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)).build();
        // Tetapkan jumlah maksimum pesan yang akan ditarik kali ini.
        int maxMessageNum = 16;
        // Tetapkan durasi tak terlihat untuk pesan.
        Duration invisibleDuration = Duration.ofSeconds(10);
        // SimpleConsumer mengharuskan klien secara aktif melakukan loop untuk mendapatkan dan memproses pesan.
        // Untuk meningkatkan konsumsi real-time, kami menyarankan menggunakan banyak thread untuk menarik pesan secara konkuren.
        while (true) {
            // Perhatikan bahwa untuk pesan dalam MessageGroup yang sama, jika pesan sebelumnya belum dikonsumsi, Anda tidak dapat memperoleh pesan berikutnya dengan memanggil Receive lagi.
            final List<MessageView> messageViewList = consumer.receive(maxMessageNum, invisibleDuration);
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // Setelah konsumsi selesai, Anda harus secara aktif memanggil ACK untuk mengirimkan hasil konsumsi.
                try {
                    consumer.ack(messageView);
                } catch (ClientException e) {
                    // Jika penarikan gagal karena pembatasan sistem atau alasan lain, Anda perlu menginisiasi ulang permintaan untuk mendapatkan pesan.
                    e.printStackTrace();
                }
            });
        }
        // Jika Anda tidak lagi memerlukan SimpleConsumer, Anda dapat mematikan proses.
        // consumer.close();
    }
}

Memperoleh log retrai konsumsi untuk pesan terurut

Retrai untuk konsumsi terurut oleh PushConsumer terjadi di klien konsumen. Server tidak dapat memperoleh log rinci untuk retrai konsumsi. Jika hasil pengiriman untuk pesan terurut dalam jejak pesan adalah 'gagal', Anda dapat memeriksa log klien konsumen untuk informasi seperti jumlah maksimum retrai dan klien konsumen.

Untuk informasi tentang jalur 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 untuk mencegah pemrosesan tidak berurutan

Kami menyarankan agar Anda mengonsumsi pesan secara serial, bukan dalam batch. Konsumsi batch dapat menyebabkan pesan diproses tidak berurutan.

Misalnya, pesan dikirim dalam urutan 1, 2, 3, dan 4. Selama konsumsi batch, urutan konsumsi mungkin 1, lalu batch [2, 3] yang gagal, lalu retrai [2, 3], dan akhirnya 4. Dalam kasus ini, kegagalan pesan 3 dapat menyebabkan pesan 2 diproses berulang kali, sehingga mengakibatkan konsumsi tidak berurutan.

Distribusikan kelompok pesan untuk menghindari hot spot

ApsaraMQ for RocketMQ memastikan bahwa pesan dalam kelompok pesan yang sama disimpan dalam antrian yang sama. Jika pesan dari skenario bisnis berbeda terkonsentrasi dalam beberapa kelompok pesan, atau bahkan satu kelompok pesan saja, tekanan penyimpanan terfokus pada beberapa antrian, atau satu antrian saja, di server. Hal ini dapat menciptakan hot spot kinerja dan menghambat skalabilitas. Praktik desain umum adalah menggunakan ID pesanan atau ID pengguna sebagai dasar pengurutan. Hal ini memastikan bahwa pesan untuk pengguna akhir yang sama diproses berurutan, sementara pesan untuk pengguna berbeda tidak perlu diproses berurutan.

Oleh karena itu, kami menyarankan agar Anda menentukan kelompok pesan dengan granularitas halus. Misalnya, Anda dapat menggunakan ID pesanan atau ID pengguna sebagai kunci kelompok pesan. Hal ini memastikan bahwa pesan yang terkait dengan pengguna akhir yang sama diproses berurutan, sementara memungkinkan pesan untuk pengguna berbeda diproses secara paralel.