ApsaraMQ for RabbitMQ:Catatan penggunaan untuk klien RabbitMQ open-source

Aktifkan pemulihan koneksi otomatis

Aktifkan pemulihan koneksi otomatis pada com.rabbitmq.client.ConnectionFactory. Tanpa fitur ini, peningkatan broker akan mengganggu operasi baca dan tulis pesan, serta klien tidak dapat terhubung ulang secara mandiri.

// Aktifkan pemulihan koneksi otomatis
factory.setAutomaticRecoveryEnabled(true);
// Atur interval antar upaya pemulihan (dalam milidetik)
factory.setNetworkRecoveryInterval(5000);

Gunakan koneksi persisten

Jangan membuka dan menutup koneksi untuk setiap operasi publish atau consume. Koneksi ulang yang sering memboroskan sumber daya jaringan dan broker, serta dapat memicu perlindungan terhadap Serangan SYN flood.

Untuk informasi lebih lanjut, lihat Koneksi dan channel.

Hindari operasi metadata dalam kode aplikasi

Operasi metadata seperti queueDeclare dan exchangeDeclare tunduk pada Pembatasan kecepatan. Jika dipicu, broker dapat menutup koneksi.

Buat antrian dan exchange melalui Konsol ApsaraMQ for RabbitMQ, bukan mendeklarasikannya saat waktu proses selama pengiriman pesan. Untuk informasi lebih lanjut tentang batas throttling, lihat Batas.

Konfigurasikan publisher confirms

Publisher confirms memungkinkan broker mengonfirmasi setiap pesan setelah ditulis ke disk, sehingga klien dapat mendeteksi dan mencoba ulang pengiriman yang gagal. Mengaktifkan publisher confirms mengubah mode pengiriman dari sinkron menjadi asinkron, yang dapat meningkatkan latensi karena broker hanya mengembalikan respons sukses setelah memastikan pesan telah dipersisten.

Untuk mengaktifkan publisher confirms pada sebuah channel:

1. Panggil confirmSelect() saat membuat channel.

   // Buat koneksi dan channel
   Connection connection = createConnection(hostName, userName, passWord, virtualHost);
   Channel channel = connection.createChannel();
   // Aktifkan publisher confirms
   channel.confirmSelect();

2. Setelah memanggil basicPublish, tunggu respons dari broker dan tangani hasilnya.

   // Publikasikan pesan (gunakan kembali channel yang ada)
   channel.basicPublish(exchangeName, bindingKey, true, props,
                   ("example body").getBytes(StandardCharsets.UTF_8));
   
   // Tunggu hingga 3 detik untuk konfirmasi
   if (channel.waitForConfirms(3000)) {
       // Pesan berhasil dipersisten ke disk
   } else {
       // Konfirmasi gagal -- kirim ulang pesan
   }

Tangani pesan yang tidak dapat dirutekan dengan flag mandatory

Mengatur parameter mandatory ke true menyebabkan broker memanggil callback ReturnListener ketika pesan tidak dapat dirutekan ke antrian mana pun.

// Daftarkan callback untuk pesan yang tidak dapat dirutekan
channel.addReturnListener(returnMessage ->
    System.out.println("return msgId=" + returnMessage.getProperties().getMessageId()));
// Publikasikan dengan mandatory=true (parameter ketiga)
channel.basicPublish(exchangeName, routingKey, true, props,
    content.getBytes(StandardCharsets.UTF_8));

Tetapkan ID pesan untuk setiap pesan

Tetapkan ID kustom untuk setiap pesan agar Anda dapat melakukan kueri jejaknya dan memecahkan masalah pengiriman. Untuk informasi lebih lanjut, lihat Bagaimana cara menentukan ID pesan?

Tangani error publikasi

Tentukan strategi penanganan error berdasarkan jenis error yang dikembalikan oleh basicPublish:

• Exception bisnis (seperti ExchangeNotExist): Lemparkan exception tersebut segera.

• Pembatasan kecepatan atau penutupan channel: Tutup koneksi yang ada, buat dan inisialisasi channel baru, lalu coba kirim ulang.

Contoh:

private void doSend(String content) throws Exception {
    try {
        // Tetapkan ID pesan unik
        String msgId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();

        channel.basicPublish(exchangeName, routingKey, true, props, content.getBytes(StandardCharsets.UTF_8));
        // Tunggu hingga 3 detik untuk konfirmasi
        if (channel.waitForConfirms(3000)) {
            // Pesan berhasil dipersisten ke disk
        } else {
            // Konfirmasi gagal -- kirim ulang pesan
        }

    } catch (Exception e) {
        String message = e.getMessage();
        System.out.println(message);

        if (channelClosedByServer(message)) {
            // Pulihkan: tutup channel lama, buat dan inisialisasi ulang channel baru, lalu coba kirim ulang
            factory.closeCon(channel);
            channel = factory.createChannel();
            this.initChannel();
            doSend(content);
        } else {
            // Error non-broker -- lemparkan exception
            throw e;
        }
    }
}

private boolean channelClosedByServer(String errorMsg) {
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        return true;
    } else {
        return false;
    }
}

Untuk kode contoh lengkap, lihat Produce messages.

Konfigurasikan mode koneksi untuk klien Spring Boot

Saat menggunakan klien Spring Boot, Anda harus secara eksplisit mengatur mode cache koneksi ke connection alih-alih mengandalkan mode default channel. Prinsip yang sama berlaku untuk pustaka klien lain yang mendukung caching koneksi — selalu konfigurasikan mode koneksi secara eksplisit.

Sebagai contoh, atur properti berikut di application.properties atau application.yml:

spring.rabbitmq.cache.connection.mode=connection

Sebagai contoh, atur properti berikut di application.properties atau application.yml:

spring.rabbitmq.cache.connection.mode=connection

Buat beberapa koneksi per produsen

Saat memproduksi pesan, setiap klien produsen harus membuat beberapa koneksi untuk terhubung ke beberapa node layanan backend. Hal ini menghindari bottleneck sumber daya akibat lonjakan lalu lintas mendadak pada satu koneksi.

Rekomendasi:

• Setiap klien produsen harus membuat 50 koneksi ke backend.

• Koneksi-koneksi tersebut harus didistribusikan ke beberapa node layanan backend.

Cegah consumption skew

Pastikan Paket didistribusikan secara merata ke seluruh konsumen. Ketidakseimbangan konsumsi menyebabkan beberapa konsumen mengalami kelebihan beban sementara yang lain menganggur. Untuk detail konfigurasi, lihat bagian "Catatan penggunaan" pada Koneksi dan saluran.

Gunakan tag konsumen yang dihasilkan broker

Biarkan broker menghasilkan tag konsumen yang unik secara global alih-alih menentukannya secara manual. Jika Anda menentukan tag konsumen kustom, pastikan tag tersebut unik di seluruh konsumen.

Konfigurasikan QoS (prefetch)

Panggil basicQos untuk membatasi jumlah pesan yang belum diacknowledge yang didorong broker ke konsumen. Ketika batas ini tercapai, broker akan menghentikan pengiriman hingga konsumen mengonfirmasi pesan-pesan yang tertunda.

Opsi cakupan:

Batasan platform:

• Nilai QoS default adalah 100 per konsumen, setara dengan channel.basicQos(100, false).

• Nilai QoS kustom tidak boleh melebihi 100. Nilai di atas 100 akan diabaikan dan nilai default yang digunakan.

• basicQos tidak berpengaruh dalam mode autoACK.

Memilih nilai QoS:

Timeout ACK dan perilaku retry

Jika konsumen tidak mengonfirmasi pesan dalam batas waktu yang dikonfigurasi, broker akan mengirim ulang pesan tersebut.

Setiap pesan dapat dicoba ulang hingga jumlah maksimum yang dikonfigurasi. Setelah semua percobaan habis, pesan akan dibuang atau diarahkan ke dead-letter exchange.

Batas waktu dan jumlah maksimum percobaan tergantung pada tipe instans dan edisi:

Gunakan basicConsume alih-alih basicGet

basicGet melakukan polling untuk pesan individual dan memiliki throughput serta batas transaksi per detik (TPS) yang lebih rendah dibandingkan basicConsume. Di lingkungan produksi dengan volume pesan tinggi, gunakan basicConsume dengan konsumen berumur panjang untuk pengiriman berbasis dorong (push-based).