Learn RabbitMQ - RPC Pattern with RabbitMQ

#Introduction

So far, every pattern we've learned is fire-and-forget: the producer sends a message and doesn't care about the result. But there are times when you need an answer. For example, asking a service to validate a credit card number and return the result — without waiting for the answer, the business flow can't continue.

This is where the RPC (Remote Procedure Call) pattern on RabbitMQ comes in. The client sends a request containing the identity of the reply queue, the server processes it, then sends a response back to that queue. The key that connects the request and response is the correlation id — a unique ID carried round-trip.

This episode shows a complete implementation of the request-reply pattern in RabbitMQ, the Direct Reply-To feature that avoids creating a queue per request, and, just as important: when you should not use RPC on RabbitMQ.

#Request-Reply Pattern

#Basic RPC Architecture

There are two actors in this pattern. The client creates a callback queue (temporary), marks the message with reply_to (the callback queue name) and correlation_id (a unique ID), then publishes to the request queue. The server takes the request, processes it, and publishes the response to the reply_to queue with the same correlation_id.

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

#Implementing the RPC Server

Here's a server that handles multiplication requests:

RPC server (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()

Notice that the server reads props.reply_to and props.correlation_id from the message properties, then sends the answer to the reply queue with the same correlation id.

#Implementing the RPC Client

The client creates a callback queue, sends the request, then waits for a response whose correlation id matches:

RPC client (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 and Timeout

#Why the Correlation ID Matters

A client can send many requests at once and wait for many responses on a single callback queue. Without a correlation id, the client can't tell which response belongs to which request. With a correlation id, every response can be paired with its originating request. The server must not change the correlation id it receives.

#Timeout Handling

The client must have a waiting time limit. If the server dies or is slow, the client shouldn't wait forever. In pika, use a deadline-based approach:

Simple timeout with a 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

The connection.process_data_events call processes incoming messages without blocking forever, so a timeout can be applied.

#Direct Reply-To

#The amq.rabbitmq.reply-to Feature

Creating a callback queue per request is expensive. RabbitMQ provides Direct Reply-To: just set reply_to to the special value amq.rabbitmq.reply-to, and the broker creates a temporary reply queue directly connected to the client's connection — with no queue declaration at all.

Use Direct Reply-To

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

With this feature, the client just waits for a message on the special default consumer without creating a callback queue. The result is faster and more resource-efficient, with the requirement that a client may only have one outstanding request at a time.

#When Not to Use RPC in RabbitMQ

#Architectural Considerations

The RPC pattern turns RabbitMQ into a synchronous request-response session, so the async advantage is lost. Avoid RPC for:

• Synchronous communication demanding very low latency — use HTTP/gRPC directly.
• Massive request-response loads with high throughput — Kafka or gRPC are a better fit.
• Flows that could actually be designed event-driven without waiting for an answer.

Use RPC in RabbitMQ only when you already have broker infrastructure, or when needs like request queuing and horizontal worker scaling are genuinely required.

#Conclusion

In episode 8 you built the request-reply pattern on RabbitMQ: RPC client and server with correlation ids and reply-to queues, timeout handling, and the Direct Reply-To amq.rabbitmq.reply-to feature to avoid creating a callback queue per request.

Key takeaways:

• RPC in RabbitMQ uses reply_to as the answer address.
• correlation_id pairs responses with requests.
• The server must return the same correlation id.
• The client needs a timeout so it doesn't wait forever.
• Direct Reply-To removes the callback queue per request.
• RPC limits client concurrency — use it carefully.
• For low latency and high throughput, consider HTTP/gRPC.

In the next episode we will dissect message properties and headers — content_type, delivery_mode, priority, correlation_id, expiration, and all the metadata a message can carry, including strategies for handling large messages and payload compression. These are small details that often distinguish a clean messaging system!