Belajar RabbitMQ - RPC Pattern dengan RabbitMQ

#Pendahuluan

Sejauh ini semua pola yang kita pelajari bersifat fire-and-forget: producer mengirim pesan dan tidak peduli dengan hasilnya. Tapi ada kalanya kalian membutuhkan jawaban. Misalnya, meminta sebuah service memvalidasi nomor kartu kredit dan mengembalikan hasil — tanpa menunggu jawaban, alur bisnis tidak bisa dilanjutkan.

Di sinilah RPC (Remote Procedure Call) pattern di atas RabbitMQ bekerja. Client mengirim request yang berisi identitas queue jawaban, server memproses, lalu mengirim response kembali ke queue tersebut. Kunci yang menghubungkan request dan response adalah correlation id — sebuah ID unik yang dibawa pulang-pergi.

Episode ini menunjukkan implementasi lengkap pola request-reply di RabbitMQ, fitur Direct Reply-To yang menghindari pembuatan queue per-request, dan yang tidak kalah penting: kapan sebaiknya tidak memakai RPC di atas RabbitMQ.

#Request-Reply Pattern

#Arsitektur Dasar RPC

Ada dua aktor dalam pola ini. Client membuat queue callback (temporary), menandai pesan dengan reply_to (nama queue callback) dan correlation_id (ID unik), lalu publish ke queue request. Server mengambil request, memproses, dan mem-publish response ke queue reply_to dengan correlation_id yang sama.

client ──publish request──► request_queue ──► server
   ▲                            ▲                │
   │                            └── response ◄───┘
   └── callback_queue ◄──── response dengan correlation_id

#Implementasi Server RPC

Berikut server yang menangani permintaan perkalian:

Server RPC (pika)

import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.queue_declare(queue="rpc_queue")
 
def on_request(ch, method, props, body):
    n = int(body)
    response = n * n
    ch.basic_publish(
        exchange="",
        routing_key=props.reply_to,
        properties=pika.BasicProperties(correlation_id=props.correlation_id),
        body=str(response),
    )
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue="rpc_queue", on_message_callback=on_request)
print("server RPC siap")
channel.start_consuming()

Perhatikan bahwa server membaca props.reply_to dan props.correlation_id dari properti pesan, lalu mengirim jawaban ke queue reply dengan correlation id yang sama.

#Implementasi Client RPC

Client membuat queue callback, mengirim request, lalu menunggu response yang correlation id-nya cocok:

Client RPC (pika)

import pika, uuid
 
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
result = channel.queue_declare(queue="", exclusive=True)
callback_queue = result.method.queue
 
corr_id = str(uuid.uuid4())
channel.basic_publish(
    exchange="",
    routing_key="rpc_queue",
    properties=pika.BasicProperties(reply_to=callback_queue, correlation_id=corr_id),
    body=b"7",
)
print(f"request dikirim dengan correlation id {corr_id}")

#Correlation ID dan Timeout

#Mengapa Correlation ID Penting

Client bisa mengirim banyak request sekaligus dan menunggu banyak response di satu queue callback. Tanpa correlation id, client tidak tahu response mana milik request mana. Dengan correlation id, setiap response bisa dipasangkan ke request asalnya. Server tidak boleh mengubah correlation id yang diterima.

#Timeout Handling

Client harus punya batas waktu menunggu. Jika server mati atau lambat, client tidak boleh menunggu selamanya. Di pika, gunakan pendekatan berbasis deadline:

Timeout sederhana dengan deadline

import time
 
deadline = time.time() + 5
while not response_ready:
    connection.process_data_events(time_limit=1)
    if time.time() > deadline:
        print("timeout menunggu response")
        break

Pemanggilan connection.process_data_events memproses pesan masuk tanpa memblokir selamanya, sehingga timeout bisa diterapkan.

#Direct Reply-To

#Fitur amq.rabbitmq.reply-to

Membuat queue callback per request mahal. RabbitMQ menyediakan Direct Reply-To: cukup set reply_to dengan nilai khusus amq.rabbitmq.reply-to, dan broker akan membuatkan reply queue sementara yang langsung terhubung ke connection client — tanpa deklarasi queue sama sekali.

Pakai Direct Reply-To

properties=pika.BasicProperties(reply_to="amq.rabbitmq.reply-to", correlation_id=corr_id)

Dengan fitur ini, client cukup menunggu pesan pada default consumer khusus tanpa membuat queue callback. Hasilnya lebih cepat dan lebih hemat resource, dengan syarat client hanya boleh punya satu request outstanding pada satu waktu.

#Kapan Tidak Memakai RPC di RabbitMQ

#Pertimbangan Arsitektur

Pola RPC mengubah RabbitMQ menjadi sesi request-response yang sinkron, sehingga keunggulan asinkron hilang. Hindari RPC untuk:

• Komunikasi synchronous yang menuntut latensi sangat rendah — pakai HTTP/gRPC langsung.
• Beban request-response massal dengan throughput tinggi — Kafka atau gRPC lebih cocok.
• Alur yang sebenarnya bisa dirancang event-driven tanpa menunggu jawaban.

Pakai RPC di RabbitMQ hanya saat sudah terlanjur punya infrastruktur broker, atau saat kebutuhan seperti queuing request dan horizontal scaling worker benar-benar dibutuhkan.

#Penutup

Di episode 8 ini kalian sudah membangun pola request-reply di atas RabbitMQ: client dan server RPC dengan correlation id dan reply-to queue, timeout handling, serta fitur Direct Reply-To amq.rabbitmq.reply-to untuk menghindari pembuatan queue callback per request.

Inti yang harus dibawa pulang:

• RPC di RabbitMQ memakai reply_to untuk alamat jawaban.
• correlation_id memasangkan response dengan request.
• Server wajib mengembalikan correlation id yang sama.
• Client butuh timeout agar tidak menunggu selamanya.
• Direct Reply-To menghilangkan queue callback per request.
• RPC membatasi concurrency client — pakai hati-hati.
• Untuk latensi rendah dan throughput tinggi, pertimbangkan HTTP/gRPC.

Di episode 9 selanjutnya kita akan membedah message properties dan headers — content_type, delivery_mode, priority, correlation_id, expiration, dan seluruh metadata yang bisa dibawa pesan, termasuk strategi menangani pesan berukuran besar dan kompresi payload. Ini detail kecil yang sering menjadi pembeda sistem pesan yang rapi!