ApsaraMQ for RabbitMQ：Praktik terbaik untuk pengendalian aliran instans

Ambang batas pengendalian aliran TPS total instans

Ambang batas pengendalian aliran TPS SendMessage node tunggal

Ambang batas pengendalian aliran operasi API

Kode kesalahan dan pesan kesalahan

• Kode kesalahan: reply-code=530

• Pesan kesalahan: reply-text=denied for too many requests

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>
(reply-code=530, reply-text=denied for too many requests, ReqId:5FB4C999314635F952FCBFF6, ErrorHelp[dstQueue=XXX_test_queue,
srcExchange=Producer.ExchangeName,bindingKey=XXX_test_bk, http://mrw.so/6rNqO8], class-id=50, method-id=20)
    at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:516)
    at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:346)
    at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:182)
    at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:114)
    at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:672)
    at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:48)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:599)
    at java.lang.Thread.run(Thread.java:748)

Contoh kode untuk menangani kode kesalahan

Kode contoh berikut menggunakan Java:

private static final int MAX_RETRIES = 5; // Jumlah maksimum percobaan ulang.
private static final long WAIT_TIME_MS = 2000; // Waktu tunggu untuk setiap percobaan ulang dalam milidetik.

private void doAnythingWithReopenChannels(Connection connection, Channel channel) {
    try {
        // ......
        // Operasi apa pun yang akan dilakukan pada saluran saat ini.
        // Misalnya, mengirim atau mengonsumsi pesan.
        // ......

    } catch (AlreadyClosedException e) {
        String message = e.getMessage();
        if (isChannelClosed(message)) {
            // Jika saluran ditutup, tutup dan buat kembali saluran tersebut.
            channel = createChannelWithRetry(connection); 
            // Anda dapat melanjutkan operasi lain setelah terhubung kembali.
            // ......
        } else {
            throw e;
        }
    }
}

private Channel createChannelWithRetry(Connection connection) {
    for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
        try {
            return connection.createChannel();
        } catch (Exception e) {
            System.err.println("Gagal membuat saluran. Percobaan " + attempt + " dari " + MAX_RETRIES);
            // Periksa kesalahan. Jika saluran masih ditutup karena pengendalian aliran, Anda dapat menunggu lalu mencoba lagi.
            // Anda juga dapat menghapus logika percobaan ulang ini.
            if (attempt < MAX_RETRIES) {
                try {
                    Thread.sleep(WAIT_TIME_MS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // Pulihkan status interupsi.
                }
            } else {
                throw new RuntimeException("Melebihi jumlah maksimum percobaan ulang untuk membuat saluran", e);
            }
        }
    }
    throw new RuntimeException("Baris ini seharusnya tidak pernah tercapai"); // Secara teori, baris kode ini tidak akan pernah tercapai.
}

private boolean isChannelClosed(String errorMsg) {
    // Periksa apakah pesan kesalahan berisi "channel.close". Kesalahan ini menunjukkan bahwa saluran ditutup.
    // Kesalahan tersebut mungkin berisi pesan kesalahan seperti 530 dan 541.
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        System.out.println("[ChannelClosed] Detail kesalahan: " + errorMsg);
        return true;
    }
    return false;
}

Solusi untuk pengendalian aliran yang disebabkan oleh melebihi TPS total instans tunggal

• Untuk skenario pengujian atau skenario jangka pendek dengan lalu lintas puncak yang tidak pasti atau volume lalu lintas kecil, kami merekomendasikan penggunaan instans ApsaraMQ for RabbitMQ serverless. Jika Anda menggunakan instans ApsaraMQ for RabbitMQ berbasis langganan dalam skenario tersebut, kami merekomendasikan agar Anda mengaktifkan fitur elastic TPS untuk instans tersebut. Untuk informasi selengkapnya, lihat Aktifkan fitur elastic TPS untuk instans.

• Untuk skenario jangka panjang dengan lalu lintas stabil dan besar, kami merekomendasikan agar Anda meningkatkan spesifikasi TPS. Untuk informasi selengkapnya, lihat Peningkatan konfigurasi instans.

Solusi untuk pengendalian aliran yang disebabkan oleh melebihi TPS pada satu node

• Kluster ApsaraMQ for RabbitMQ menggunakan arsitektur terdistribusi. Kami merekomendasikan agar Anda membuat beberapa koneksi (minimal 10) untuk setiap antrian sehingga klien dapat terhubung ke beberapa node layanan dalam kluster secara merata. Metode ini secara efektif mencegah hotspot beban dan meningkatkan efisiensi pengiriman serta konsumsi pesan.

• Jika Anda pengguna Spring, kami merekomendasikan agar Anda menggunakan mode CONNECTION dari CachingConnectionFactory. Untuk informasi selengkapnya, lihat Integrasi Spring.