Request-Reply Pattern — Complete Guide

In this tutorial, you will learn about Request. We cover key concepts, practical examples, and best practices to help you master this topic.

The request-reply pattern enables synchronous-style request-response communication over asynchronous message queues using correlation IDs and temporary reply queues.

What You'll Learn

By the end of this lesson, you will understand the request-reply pattern, how to implement it with RabbitMQ, how correlation IDs match requests to responses, and when to use this pattern over direct HTTP calls.

Why It Matters

Some operations need a response: "Is this user valid?", "What is the exchange rate?", "Process this payment and give me the receipt." Direct HTTP calls work but couple the services. The request-reply pattern over a message queue provides decoupling with response correlation.

Real-World Use

A fraud detection service needs to check a Transaction before processing. The payment service sends a request message with order details and waits for a response. The fraud service processes it and replies with a risk score. Both services are decoupled by the queue.

Request-Reply Architecture

sequenceDiagram
    participant C as Client
    participant Q as Request Queue
    participant S as Server
    participant R as Reply Queue

    C->>Q: Send request (correlation_id, reply_to)
    Q->>S: Deliver request
    S->>C: Send reply (correlation_id)
    C->>C: Match reply by correlation_id

The client sends a request with a unique correlation ID and specifies a reply queue. The server processes the request and sends the response to the reply queue with the same correlation ID. The client matches responses by correlation ID.

Implementing Request-Reply with RabbitMQ

import pika
import json
import uuid

class RPCClient:
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True
        )
        self.response = None
        self.correlation_id = None

    def on_response(self, ch, method, properties, body):
        if self.correlation_id == properties.correlation_id:
            self.response = body

    def call(self, request_data):
        self.correlation_id = str(uuid.uuid4())
        self.response = None

        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            body=json.dumps(request_data),
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.correlation_id,
            )
        )

        while self.response is None:
            self.connection.process_data_events()
        return json.loads(self.response)

client = RPCClient()
result = client.call({'method': 'add', 'args': [5, 3]})
print(f"Result: {result}")

Expected output:

Result: 8

# RPC Server
import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue')

def on_request(ch, method, properties, body):
    request = json.loads(body)
    print(f"Received: {request}")

    if request['method'] == 'add':
        result = sum(request['args'])
    else:
        result = None

    ch.basic_publish(
        exchange='',
        routing_key=properties.reply_to,
        body=json.dumps(result),
        properties=pika.BasicProperties(correlation_id=properties.correlation_id)
    )
    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("RPC Server waiting for requests...")
channel.start_consuming()

Expected output:

RPC Server waiting for requests...
Received: {'method': 'add', 'args': [5, 3]}

Why Correlation IDs Matter

Think of a busy restaurant kitchen. Multiple waiters place orders on the same rack. When a dish is ready, the chef calls out the table number. Each waiter knows which table ordered what dish.

Correlation IDs work the same way. Multiple clients send requests. The server processes them in any order and sends each response with the original correlation ID. Each client listens for its own correlation ID and ignores responses meant for others.

Request-Reply vs HTTP

Common Mistakes

1. Forgetting the Reply Queue

The request must specify a reply_to queue. Without it, the server has no way to send the response. The client creates an exclusive queue for replies.

2. Ignoring Timeout

A request-reply client can block forever if the server crashes or the message is lost. Always set a timeout. If no response arrives within the timeout, the client should fail gracefully.

3. Using the Same Correlation ID for Multiple Requests

Each request must have a unique correlation ID. Reusing IDs causes the wrong response to be matched to a request. Use UUIDs or incrementing counters.

4. Blocking the Client Event Loop

The RPC client pattern blocks the event loop waiting for a response. In production, use asynchronous RPC or set a timeout. Blocking the event loop prevents the client from handling other tasks.

5. Not Handling Server Failures

If the server crashes after processing but before sending the response, the client never gets a response. Implement server-side idempotency and client-side retry with timeout.

Practice Questions

1. How does the request-reply pattern work over message queues?

The client sends a request with a correlation ID and a reply queue address. The server processes it and sends the response to the reply queue with the same correlation ID. The client matches responses by correlation ID.

2. What is a correlation ID?

A unique identifier that pairs a request with its response. The client generates it and includes it in the request. The server echoes it back in the response so the client knows which request the response belongs to.

3. When would you use request-reply over HTTP?

When you want the decoupling benefits of message queues (load leveling, fault tolerance, at-least-once delivery) while still needing a response. Use HTTP when simplicity and lower latency are more important.

4. How do you handle timeout in request-reply?

Set a timeout on the client side. If no response arrives within the timeout, the client can retry the request or fail. Use a message TTL to prevent abandoned requests from accumulating.

Challenge

Design a distributed calculator service using request-reply. Clients send calculation requests (add, multiply, sqrt). Multiple server instances handle requests. Implement timeout, retry, and Load Balancing across servers.

FAQ

Mini Project: RPC Calculator

import pika
import json
import uuid

class CalculatorServer:
    def __init__(self):
        self.conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.ch = self.conn.channel()
        self.ch.queue_declare(queue='calc_requests')
        self.ch.basic_consume(queue='calc_requests', on_message_callback=self.handle)

    def handle(self, ch, method, properties, body):
        req = json.loads(body)
        op = req['op']
        a, b = req['a'], req['b']

        if op == 'add': result = a + b
        elif op == 'sub': result = a - b
        elif op == 'mul': result = a * b
        elif op == 'div': result = a / b if b != 0 else 'error'

        ch.basic_publish(
            exchange='', routing_key=properties.reply_to,
            body=json.dumps(result),
            properties=pika.BasicProperties(correlation_id=properties.correlation_id)
        )
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(f"Calculated: {a} {op} {b} = {result}")

    def start(self):
        self.ch.start_consuming()

import threading
server = CalculatorServer()
t = threading.Thread(target=server.start, daemon=True)
t.start()

Expected output:

Calculated: 5 + 3 = 8

What's Next

Now that you understand request-reply, explore competing consumers for distributing work across multiple workers, then learn about fanout exchanges for broadcast messaging.