Request-Reply (RPC) Pattern
IntermediateProducer sends a message with a reply-to queue and correlation-id; consumer processes the request and publishes the response to the reply-to queue; producer correlates the reply.
Overview
Messaging is inherently one-way (fire-and-forget), but many workflows need a response — pricing a basket, querying an inventory service, calling a remote procedure. The RabbitMQ RPC pattern simulates request-reply over AMQP: the client publishes a request to a well-known queue with two AMQP properties: `reply_to` (the name of a temporary, exclusive queue the client is listening on) and `correlation_id` (a UUID the client uses to match the response). The server consumes the request, processes it, and publishes the result directly to `reply_to`. Because multiple requests may be in-flight simultaneously, the `correlation_id` is essential for matching responses to pending futures. Spring AMQP's `RabbitTemplate.convertSendAndReceive()` encapsulates this pattern entirely.
Manual RPC with AMQP Java Client
Create an exclusive auto-delete reply queue, store a map of correlationId → CompletableFuture, publish with `replyTo` and `correlationId` properties, and complete the future when the response arrives on the reply queue.
// Client side
String replyQueue = channel.queueDeclare("", false, true, true, null).getQueue();
Map<String, CompletableFuture<byte[]>> pending = new ConcurrentHashMap<>();
channel.basicConsume(replyQueue, true, (tag, msg) -> {
String corrId = msg.getProperties().getCorrelationId();
CompletableFuture<byte[]> future = pending.remove(corrId);
if (future != null) future.complete(msg.getBody());
}, tag -> {});
// Send request
String corrId = UUID.randomUUID().toString();
CompletableFuture<byte[]> result = new CompletableFuture<>();
pending.put(corrId, result);
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.correlationId(corrId)
.replyTo(replyQueue)
.build();
channel.basicPublish("", "rpc_queue", props, "42".getBytes());
byte[] response = result.get(5, TimeUnit.SECONDS);
// Server side
channel.basicConsume("rpc_queue", false, (tag, msg) -> {
byte[] body = processRequest(msg.getBody());
AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
.correlationId(msg.getProperties().getCorrelationId())
.build();
channel.basicPublish("", msg.getProperties().getReplyTo(), replyProps, body);
channel.basicAck(msg.getEnvelope().getDeliveryTag(), false);
}, tag -> {});Spring AMQP — RabbitTemplate RPC
`RabbitTemplate.convertSendAndReceive()` handles the temporary reply queue and correlation ID automatically via the `DirectReplyToMessageListenerContainer`. No need to create a reply queue manually. It blocks the calling thread until the reply arrives or the `replyTimeout` expires.
@Configuration
class RpcConfig {
@Bean
RabbitTemplate rabbitTemplate(ConnectionFactory cf) {
var t = new RabbitTemplate(cf);
t.setReplyTimeout(5_000); // ms
return t;
}
@Bean
Queue rpcQueue() { return new Queue("rpc_queue"); }
}
// Client
@Service
class PricingClient {
private final RabbitTemplate rabbit;
public String price(String itemId) {
// blocks up to replyTimeout; returns null on timeout
return (String) rabbit.convertSendAndReceive("rpc_queue", itemId);
}
}
// Server
@RabbitListener(queues = "rpc_queue")
public String handlePricingRequest(String itemId) {
return priceService.price(itemId); // return value → reply-to queue
}RPC Pitfalls and Alternatives
Synchronous RPC over messaging inherits the worst of both worlds: the latency of messaging with the coupling of synchronous calls. Prefer async patterns (callbacks, futures, event notifications) when possible. If you must use RPC, always set a timeout, handle null responses (timeout), and consider using HTTP/gRPC for truly synchronous internal calls instead. Also, exclusive reply queues create one connection per client; prefer RabbitMQ's pseudo-queue `amq.rabbitmq.reply-to` (Direct Reply-To) to avoid overhead.
# Direct Reply-To — no explicit queue needed (RabbitMQ >= 3.4)
# Use the pseudo-queue "amq.rabbitmq.reply-to" as replyTo
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.correlationId(corrId)
.replyTo("amq.rabbitmq.reply-to") // magic value
.build();
channel.basicPublish("", "rpc_queue", props, request);
// Consume from the same pseudo-queue (auto-ack forced)
channel.basicConsume("amq.rabbitmq.reply-to", true, replyHandler, cancelHandler);Key Points to Remember
- 1RPC over AMQP uses replyTo (temporary queue) and correlationId to match requests to responses
- 2Spring AMQP's convertSendAndReceive() handles reply-to queue creation and correlation automatically
- 3Direct Reply-To (amq.rabbitmq.reply-to) avoids creating a new queue per client
- 4Always set a replyTimeout — a missing or crashed server will never send a reply
- 5Synchronous RPC over messaging couples services; prefer async callbacks for resilience
- 6The server must publish the response with the same correlationId from the request
Interview Questions
Sign in to ask AriaWhat are the roles of replyTo and correlationId in the RabbitMQ RPC pattern?
What is Direct Reply-To and how does it differ from creating a temporary reply queue?
What are the drawbacks of implementing RPC over a message broker instead of HTTP?
How does Spring AMQP's RabbitTemplate manage in-flight RPC correlations?
What happens if the RPC server crashes before publishing the reply?
Ask Aria about Request-Reply (RPC) Pattern
Your personal AI tutor — ask anything about this concept
Revision Status
Personal Notes
Sign in to save personal notes for this topic.
Discussion
Sign in to join the discussion.