Implementing RPC client and server in RabbitMQ - Commands & Configuration

Introduction

Sometimes, one program needs to ask another program to do a task and wait for the answer. This is called Remote Procedure Call (RPC). RabbitMQ helps programs talk this way by sending messages back and forth safely and quickly.

When a web app needs to ask a backend service to process data and wait for the result before continuing.

When a microservice needs to request information from another microservice and use the response immediately.

When you want to split a big task into smaller parts handled by different programs and collect their answers.

When you want to build a system where clients send requests and servers reply asynchronously but reliably.

When you want to avoid blocking your program while waiting by using message queues for communication.

Config File - rpc_server.py

rpc_server.py

import pika
import time

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)

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)
    print(f" [.] fib({n})")
    response = fib(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(" [x] Awaiting RPC requests")
channel.start_consuming()

This Python script sets up the RPC server using RabbitMQ.

• queue_declare creates a queue named rpc_queue where requests arrive.
• on_request is the function called when a request message arrives; it calculates the Fibonacci number.
• The server sends the result back to the client using the reply_to queue and correlation_id to match requests and responses.
• basic_qos limits the server to process one request at a time for fairness.

Commands

Start the RPC server. It waits for requests and processes them one by one.

Terminal

python rpc_server.py

Expected Output

 [x] Awaiting RPC requests

Run the RPC client script that sends a request to calculate a Fibonacci number and waits for the answer.

Terminal

python rpc_client.py

Expected Output

 [x] Requesting fib(7)

 [.] Got fib(7) = 13

Key Concept

RPC with RabbitMQ uses a request queue for clients to send tasks and reply queues with correlation IDs to receive matching responses asynchronously.

Common Mistakes

Not setting the reply_to property in the request message

The server won't know where to send the response, so the client never receives the answer.

Always set the reply_to property to a unique callback queue name in the client request.

Not using correlation_id to match responses

If multiple requests are sent, the client can't tell which response belongs to which request.

Use a unique correlation_id for each request and check it when receiving responses.

Not acknowledging messages on the server

Messages may be redelivered or lost, causing duplicate processing or message loss.

Call basic_ack after processing each request to confirm successful handling.

Summary

Start the RPC server to listen on a queue for incoming requests.

Send requests from the client with reply_to and correlation_id properties.

The server processes requests and sends responses back to the client's callback queue.

Clients wait for responses and match them using correlation_id to handle multiple requests safely.

Practice

1. What is the main role of the RPC server in RabbitMQ RPC communication?

easy

A. To store messages permanently without processing

B. To send requests and wait for responses

C. To receive requests, process them, and send back responses

D. To monitor network traffic between clients

Solution

1. Step 1: Understand RPC server role

   The RPC server listens for requests on a queue and processes them.

2. Step 2: Identify server response behavior

   After processing, the server sends the result back to the client via a reply queue.

3. Final Answer:

   To receive requests, process them, and send back responses -> Option C

4. Quick Check:

   RPC server handles requests and replies [OK]

Hint: Server processes requests and replies, client sends requests [OK]

Common Mistakes:

• Confusing client and server roles
• Thinking server only stores messages
• Assuming server sends requests

2. Which of the following is the correct way to declare a callback queue for RPC replies in Python pika client?

easy

A. channel.queue_declare(queue='rpc_queue')

B. result = channel.queue_declare(queue='', exclusive=True)

C. channel.basic_publish(exchange='', routing_key='rpc_queue')

D. channel.basic_consume(queue='rpc_queue', on_message_callback=callback)

Solution

1. Step 1: Identify callback queue declaration

   For RPC replies, a temporary exclusive queue is declared with empty name and exclusive=True.

2. Step 2: Match syntax with pika usage

   Using queue_declare with queue='' and exclusive=True creates a unique callback queue.

3. Final Answer:

   result = channel.queue_declare(queue='', exclusive=True) -> Option B

4. Quick Check:

   Exclusive empty queue for callback = result = channel.queue_declare(queue='', exclusive=True) [OK]

Hint: Use empty queue name with exclusive=True for callback queue [OK]

Common Mistakes:

• Declaring fixed queue name for callback
• Using basic_publish instead of queue_declare
• Not setting exclusive=True for callback queue

3. Given this Python RPC client snippet, what will be printed?

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='5'
)
print('Sent request with correlation_id:', corr_id)

medium

A. Sent request with correlation_id:

B. Sent request with correlation_id: rpc_queue

C. Sent request with correlation_id: 5

D. Error: correlation_id not defined

Solution

1. Step 1: Understand correlation_id assignment

   corr_id is assigned a new UUID string using uuid.uuid4(), which is random.

2. Step 2: Analyze print statement output

   The print outputs the fixed string plus the generated corr_id value.

3. Final Answer:

   Sent request with correlation_id: <random UUID> -> Option A

4. Quick Check:

   corr_id is UUID string printed [OK]

Hint: corr_id is a UUID string, printed as is [OK]

Common Mistakes:

• Confusing correlation_id with routing_key
• Printing message body instead of corr_id
• Assuming correlation_id is undefined

4. In this RPC server code snippet, what is the main error?

def on_request(ch, method, props, body):
    n = int(body)
    response = fib(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_consume(queue='rpc_queue', on_message_callback=on_request)

medium

A. Missing call to start consuming messages

B. Incorrect use of correlation_id property

C. No acknowledgment of message delivery

D. Using wrong queue name in basic_consume

Solution

1. Step 1: Check message consumption setup

   basic_consume is set but start_consuming() is not called to begin processing.

2. Step 2: Verify other parts

   Correlation_id usage and ack are correct; queue name matches.

3. Final Answer:

   Missing call to start consuming messages -> Option A

4. Quick Check:

   start_consuming() needed to process messages [OK]

Hint: Always call start_consuming() after basic_consume [OK]

Common Mistakes:

• Forgetting start_consuming() call
• Misusing correlation_id property
• Not acknowledging messages

5. You want to implement an RPC client that sends multiple requests concurrently and matches replies correctly. Which approach best ensures correct matching of replies to requests?

hard

A. Create a new reply queue for each request without correlation_id

B. Send all requests with the same correlation_id to simplify matching

C. Use a single shared reply queue without correlation_id

D. Use a unique correlation_id for each request and check it on reply

Solution

1. Step 1: Understand concurrency in RPC client

   Multiple requests need unique IDs to identify replies correctly.

2. Step 2: Evaluate options for reply matching

   Using unique correlation_id per request allows client to match replies properly.

3. Final Answer:

   Use a unique correlation_id for each request and check it on reply -> Option D

4. Quick Check:

   Unique correlation_id matches replies to requests [OK]

Hint: Assign unique correlation_id per request for reply matching [OK]

Common Mistakes:

• Using same correlation_id for all requests
• Not using correlation_id at all
• Creating many reply queues unnecessarily