ApsaraMQ for RocketMQ:Model topik Lite

Nama topik Lite

• Definisi: Nama yang mengidentifikasi lite topic. Nama lite topic bersifat unik secara global dalam cakupan topik induknya.

• Nilai: Saat tipe topik adalah Lite dan Anda memanggil setLiteTopic pada sebuah pesan, sistem akan secara otomatis membuat lite topic tersebut jika belum ada.

• Batasan: Lihat Batasan parameter.

Time-to-live (TTL)

• Definisi: Waktu kedaluwarsa untuk lite topic. Jika tidak ada pesan baru yang ditulis ke lite topic selama periode yang melebihi TTL-nya, sistem akan menghapusnya secara otomatis. Penghapusan berarti melepaskan kuota yang dialokasikan untuk lite topic tersebut (jumlah total dikurangi satu).

• Nilai: Saat membuat topik bertipe Lite, Anda dapat mengatur parameter kedaluwarsa.

• Batasan: Lihat Batasan parameter.

Kasus penggunaan 1: Komunikasi asinkron untuk sistem Multi-Agent guna mengatasi pemblokiran panggilan berdurasi panjang

Saat skenario AI semakin kompleks, sistem single-agent menghadapi keterbatasan: kurang spesialisasi, kesulitan mengintegrasikan berbagai domain, dan ketidakmampuan mendukung pengambilan keputusan kolaboratif dinamis. Aplikasi dan alur kerja single-agent beralih ke arsitektur Multi-Agent. Namun, karena tugas AI sering kali memakan waktu lama, panggilan sinkron memblokir thread pemanggil, sehingga membatasi skalabilitas untuk kolaborasi skala besar.

Seperti yang ditunjukkan di atas, alur kerja Multi-Agent bekerja sebagai berikut: Supervisor Agent membagi permintaan menjadi dua subtugas untuk dua child agent. Setiap child agent menyelesaikan bagiannya dan mengembalikan hasil ke Supervisor Agent, yang kemudian menggabungkannya dan mengirim respons akhir ke klien web. Dengan menggunakan RocketMQ untuk komunikasi asinkron:

1. Alur penanganan permintaan:

   1. Buat topik (Request) untuk setiap child agent sebagai antrian buffer tugas. Gunakan topik prioritas untuk memproses tugas prioritas tinggi terlebih dahulu.

   2. Supervisor Agent mengirim detail subtugas ke topik request yang sesuai.

2. Alur penanganan respons:

   1. Supervisor Agent membuat topik bertipe Lite (Response) dan berlangganan kepadanya.

   2. Setiap child agent mengirim hasil tugasnya ke LiteTopic di bawah topik Response. Beri nama setiap LiteTopic menggunakan ID tugas agar setiap tugas memiliki LiteTopic khusus sendiri.

   3. Supervisor Agent menerima hasil secara real time melalui langganan dan mendorongnya ke klien web menggunakan HTTP SSE.

Kasus penggunaan 2: Pengelolaan status sesi terdistribusi untuk mengatasi masalah kontinuitas sesi dalam aplikasi AI

Interaksi aplikasi AI bersifat unik: berdurasi panjang, multi-turn, dan sangat bergantung pada sumber daya komputasi mahal per sesi. Saat aplikasi mengandalkan koneksi persisten seperti SSE, gangguan apa pun (akibat restart gerbang, timeout, atau ketidakstabilan jaringan) menyebabkan hilangnya konteks sesi saat ini dan membuang sumber daya komputasi AI yang telah diinvestasikan.

Seperti pada alur respons kasus penggunaan 1, gunakan topik bertipe Lite untuk notifikasi hasil real time. Beri nama setiap LiteTopic menggunakan SessionID (misalnya, chatbot/{sessionID}). Semua hasil sesi dikirim sebagai pesan terurut dalam topik ini. Untuk menjaga kontinuitas sesi setelah koneksi ulang:

1. Klien web membentuk koneksi persisten dengan node server aplikasi 1 dan memulai sesi Session2.

2. Node server aplikasi 1 berlangganan ke LiteTopic [chat/SessionID2].

3. Penjadwal tugas Large Language Model (LLM) mengirim hasil ke LiteTopic [chat/SessionID2] berdasarkan SessionID dalam permintaan.

4. Karena masalah jaringan, WebSocket terhubung ulang ke node server aplikasi 2.

5. Node server aplikasi 1 berhenti berlangganan dari LiteTopic [chat/SessionID2]. Node server aplikasi 2 berlangganan kepadanya.

6. LiteTopic [chat/SessionID2] melanjutkan pengiriman dari offset terakhir yang dikonsumsi, memastikan kontinuitas status dan data sesi.

Kirim pesan

Producer producer = provider.newProducerBuilder()
    .setTopics(topic)
    .setClientConfiguration(clientConfiguration)
    .build();
final Message message = provider.newMessageBuilder()
    .setTopic(topic)
    // Setel kunci pesan untuk pencarian tepat berdasarkan kata kunci.
    .setKeys("messageKey")
    // Setel LiteTopic
    .setLiteTopic("lite-topic-1")
    // Isi pesan
    .setBody("messageBody".getBytes())
    .build();
try {
    final SendReceipt sendReceipt = producer.send(message);
    log.info("Pesan berhasil dikirim, messageId={}", sendReceipt.getMessageId());
} catch (LiteTopicQuotaExceededException e) {
    // Kuota LiteTopic terlampaui. Evaluasi dan tingkatkan kuota.
    log.error("Kuota lite topic terlampaui", e);
} catch (Throwable t) {
    log.error("Gagal mengirim pesan", t);
}

Konsumsi pesan

Gunakan kelas LitePushConsumer:

// Inisialisasi LitePushConsumer dengan kelompok konsumen, topik target, dan parameter komunikasi.
LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
    .setClientConfiguration(clientConfiguration)
    // Topik yang diikat ke ConsumerGroup saat dibuat di Konsol
    .bindTopic(topicName)
    // Setel kelompok konsumen
    .setConsumerGroup(consumerGroup)
    .setMessageListener(messageView -> {
        // Proses pesan dan kembalikan hasil konsumsi.
        LOGGER.info("Konsumsi pesan={}", messageView);
        return ConsumeResult.SUCCESS;
    })
    .build();
try {
    // Berlangganan ke LiteTopic yang diinginkan
    litePushConsumer.subscribeLite("lite-topic-1");
    litePushConsumer.subscribeLite("lite-topic-2");
    litePushConsumer.subscribeLite("lite-topic-3");
} catch (LiteSubscriptionQuotaExceededException e) {
    // Kuota langganan LiteTopic terlampaui. Evaluasi dan tingkatkan kuota.
    log.error("Kuota langganan lite terlampaui", e);
} catch (Throwable t) {
    log.error("Gagal berlangganan lite topic", t);
}
// Setelah pemrosesan bisnis selesai, segera berhenti berlangganan dari LiteTopic yang tidak digunakan
litePushConsumer.unsubscribeLite("lite-topic-3");
// Dapatkan himpunan LiteTopic yang sedang dilanggan
Set<String> liteTopicSet = litePushConsumer.getLiteTopicSet();

Perbarui langganan secara dinamis

/**
 * Tambahkan langganan secara dinamis.
 * Metode subscribeLite() melakukan panggilan jaringan dan memvalidasi kuota,
 * sehingga bisa gagal.
 * Selalu periksa hasilnya untuk memastikan langganan berhasil.
 * Kemungkinan skenario kegagalan:
 * 1. Kesalahan jaringan – ulangi panggilan.
 * 2. Validasi kuota gagal – melempar LiteSubscriptionQuotaExceededException.
 * Evaluasi apakah kuota Anda memenuhi persyaratan, dan segera panggil
 * unsubscribeLite() untuk melepaskan sumber daya dari topik yang tidak digunakan.
 */
litePushConsumer.subscribeLite("lite-topic-1");
// Hapus langganan secara dinamis
litePushConsumer.unsubscribeLite("lite-topic-1");

Instans Serverless

Instans non-Serverless (subscription dan pay-as-you-go)