ApsaraMQ for RabbitMQ:Integrasi Spring

Praktik Terbaik Koneksi dan Channel

ApsaraMQ for RabbitMQ menggunakan arsitektur multi-node terdistribusi di backend. Untuk menghindari bottleneck performa pada satu node di sisi server, ikuti praktik terbaik berikut saat mengonfigurasi koneksi klien.

• Gunakan mode CONNECTION: Konfigurasikan klien untuk membuat beberapa koneksi TCP ke server. Dalam mode CONNECTION, satu klien mendistribusikan koneksi ke beberapa node layanan backend, sehingga memanfaatkan resource dari banyak node dalam kluster. Disarankan untuk mengonfigurasi jumlah koneksi (CONNECTION) per klien pada level puluhan, disesuaikan dengan volume traffic klien.

• Hindari pembuatan dan pemutusan koneksi yang sering: Jangan sering membuat dan menutup koneksi. Siklus koneksi yang sering mengurangi efisiensi pengiriman dan konsumsi pesan. Pertahankan koneksi persisten alih-alih membuat koneksi untuk setiap operasi.

• Batasi jumlah channel per koneksi: Atur jumlah channel yang dapat dibuat pada satu koneksi tidak lebih dari 50. Terlalu banyak channel pada satu koneksi dapat menyebabkan lonjakan traffic TCP mendadak. Ditambah dengan kemungkinan pengurutan ulang paket, hal ini dapat memicu mekanisme Linux Netfilter Conntrack untuk mengklasifikasikan paket sebagai INVALID, sehingga klien secara aktif mengirim RST untuk mereset koneksi.

Anda dapat mengonfigurasi mode koneksi dan jumlah CONNECTION/CHANNEL menggunakan parameter berikut:

• spring.rabbitmq.cache.connection.mode – Atur ke connection untuk mode CONNECTION.

• spring.rabbitmq.cache.connection.size – Jumlah koneksi. Disarankan: level puluhan.

• spring.rabbitmq.cache.channel.size – Jumlah channel per koneksi. Disarankan: tidak lebih dari 50.

Produksi pesan

Dalam RabbitMQService, peroleh RabbitTemplate melalui dependency injection dan panggil metode send-nya untuk mengirim pesan.

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
public class RabbitMQService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange, String routingKey, String content) {
        // Atur MessageId.
        String msgId = UUID.randomUUID().toString();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setMessageId(msgId);
        // Buat Message.
        Message message = new Message(content.getBytes(StandardCharsets.UTF_8), messageProperties);
        /*
         * Panggil antarmuka send() untuk mengirim pesan.
         * exchange: nama exchange.
         * routingKey: kunci routing.
         * message: isi pesan.
         * correlationData digunakan untuk publisher confirms.
         */
        rabbitTemplate.send(exchange, routingKey, message, null);
    }
}

Konsumsi pesan

Gunakan anotasi @RabbitListener untuk mengonsumsi pesan:

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class MessageListener {

     /**
     * Menerima pesan.
     * @param message Pesan.
     * @param channel Channel.
     * @throws IOException
     * Ganti queues dengan nama antrian yang telah Anda buat.
     */
    @RabbitListener(queues = "myQueue")
    public void receiveFromMyQueue(Message message, Channel channel) throws IOException {
        // Masukkan logika bisnis untuk konsumsi pesan.
        ...
        // Anda harus mengembalikan Ack dalam periode validitas Ack (timeout konsumsi). Jika tidak, konfirmasi tidak valid dan pesan akan dikirim ulang.
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

Berikut adalah konfigurasi opsional umum untuk RabbitListener:

Catatan penting

Verifikasi apakah klien Anda menggunakan metode receiveAndConvert dari RabbitTemplate untuk konsumsi pesan.

Metode ini menerima parameter seperti nama antrian dan durasi timeout. Saat timeout ditentukan, logika pemrosesan internal memiliki masalah berikut: sebelum mengonsumsi pesan, metode ini terlebih dahulu mengirim perintah BasicCancel untuk membatalkan langganan konsumen, lalu mengirim Ack untuk mengonfirmasi pesan.

Namun, saat ApsaraMQ for RabbitMQ menerima permintaan BasicCancel, server segera membatalkan konsumen yang bersangkutan dan memasukkan kembali semua pesan yang belum diakui ke antrian. Akibatnya, Ack yang dikirim klien setelah itu menjadi tidak valid — server tidak menganggapnya sebagai acknowledgment yang sah, melainkan mengirim ulang pesan-pesan tersebut, sehingga menyebabkan konsumsi duplikat di sisi klien.

Selain itu, jika antrian memiliki throughput pesan yang tinggi, operasi cancel dan requeue yang sering dapat menyebabkan akumulasi pesan, yang selanjutnya memengaruhi stabilitas konsumsi.