Business Problem & Why This Solution Matters

The Challenge

In modern distributed systems, we often need to:

• Send a request and wait for a specific response
• Maintain correlation between requests and responses
• Implement timeout mechanisms for reliability
• Handle multiple concurrent requests efficiently

Traditional messaging with Kafka is fire-and-forget (asynchronous), but many business scenarios require synchronous behavior:

• Financial transactions: Need immediate confirmation
• User authentication: Must wait for validation response
• Data validation: Require real-time feedback
• API gateways: Need to aggregate responses from multiple services

The Solution

This implementation leverages RabbitMQ’s native RPC pattern which provides:

• Built-in correlation ID support through message properties
• Direct response routing using replyTo queue specification
• Instance-specific response queues for multi-instance deployments
• Push-based message delivery (no polling required)
• Automatic connection management and recovery

Architecture Overview

┌─────────────────┐    request-queue    ┌─────────────────┐  
│ RabbitMQ Client │ ──────────────────► │ RabbitMQ Server │  
│                 │                     │                 │  
│ ┌─────────────┐ │  replyTo: response- │ ┌─────────────┐ │  
│ │Pending      │ │  queue-{instanceId} │ │Request      │ │  
│ │Requests     │ │                     │ │Processor    │ │  
│ │Map          │ │ correlationId: UUID │ │             │ │  
│ └─────────────┘ │                     │ └─────────────┘ │  
│                 │ ◄────────────────── │                 │  
└─────────────────┘  response-queue-    └─────────────────┘  
                     {instanceId}

Key Components

RabbitMQSyncClient

Purpose: Sends synchronous requests and waits for responses using RabbitMQ’s RPC pattern

Key Features:

• Correlation ID Management: Each request gets a unique UUID
• Instance-Specific Response Queues: Creates response-queue-{instanceId} for each client instance
• Timeout Handling: Configurable request timeout (default: 30 seconds)
• Concurrent Request Support: Uses ConcurrentHashMap to track multiple pending requests
• Native RabbitMQ RPC: Utilizes replyTo and correlationId message properties

Implementation Highlights

fun sendSyncRequest(request: String): String {  
    val correlationId: UUID = UUID.randomUUID()  
    val future = CompletableFuture<String>()  
    pendingRequests\[correlationId\] = future  
      
    // RabbitMQ RPC properties - native support for request-response  
    val props: AMQP.BasicProperties = AMQP.BasicProperties.Builder()  
        .correlationId(correlationId.toString())  
        .replyTo(responseQueue)  // Instance-specific response queue  
        .build()  
      
    // Send request with RPC properties  
    channel.basicPublish("", rabbitMQProperties.requestQueue, props, request.toByteArray())  
    log.info("Sent request to queue ${rabbitMQProperties.requestQueue} with correlationId: $correlationId")  
      
    return try {  
        // Block until response is received or timeout occurs  
        future.get(rabbitMQProperties.requestTimeout, TimeUnit.MILLISECONDS)  
    } catch (e: Exception) {  
        pendingRequests.remove(key = correlationId)  
        log.error("Request timed out or failed. Exception: ${e.localizedMessage}")  
        throw RuntimeException("Request timed out or failed", e)  
    } finally {  
        log.info("Request with correlationId: $correlationId completed")  
        future.complete(null)  
    }  
}

Response Consumer Setup (Push-based, no polling)

override fun start() {  
    // Create connection and channel  
    connection = connectionFactory.newConnection()  
    channel = connection.createChannel()  
      
    // Declare queues  
    channel.queueDeclare(rabbitMQProperties.requestQueue, true, false, false, null)  
    channel.queueDeclare(responseQueue, true, false, false, null)  
      
    // Set up push-based response consumer  
    val deliverCallback = DeliverCallback { consumerTag: String, delivery: Delivery ->  
        val correlationId: UUID? = UUID.fromString(delivery.properties.correlationId)  
        val response: String = String(bytes = delivery.body)  
        log.info("Received response for correlationId: $correlationId")  
          
        // Complete the pending request  
        val future: CompletableFuture<String>? = pendingRequests.remove(key = correlationId)  
        future?.complete(response)  
    }  
      
    // Start consuming responses (push-based delivery)  
    channel.basicConsume(responseQueue, true, deliverCallback) { consumerTag ->  
        log.info("Consumer was cancelled: $consumerTag")  
    }  
      
    running = true  
    log.info("RabbitMQ client started with instance ID: ${rabbitMQProperties.instanceId}")  
}

RabbitMQSyncServer

Purpose: Processes requests and sends responses back to the appropriate client using RPC pattern

Key Features:

• Request Processing: Listens to request-queue
• Dynamic Response Routing: Uses replyTo property to send responses to correct client queue
• Correlation ID Preservation: Maintains correlation ID through the request-response cycle
• Automatic Response Generation: Processes requests and generates responses
• Error Resilience: Handles malformed requests gracefully

override fun start() {  
    // Create connection and channel  
    connection = connectionFactory.newConnection()  
    channel = connection.createChannel()  
      
    // Declare request queue  
    channel.queueDeclare(rabbitMQProperties.requestQueue, true, false, false, null)  
      
    // Set up request consumer  
    val deliverCallback = DeliverCallback { consumerTag: String?, delivery: Delivery ->  
        val request: String = String(bytes = delivery.body)  
        val correlationId: String? = delivery.properties.correlationId  
        val replyTo: String? = delivery.properties.replyTo  
          
        log.info("Received request: $request from queue: ${rabbitMQProperties.requestQueue}")  
          
        if (correlationId != null && replyTo != null) {  
            // Process business logic  
            val processedResponse = processBusinessLogic(request, correlationId)  
              
            // Send response using RPC pattern  
            sendResponse(replyTo, correlationId, processedResponse)  
        } else {  
            log.error("Received request without correlationId or replyTo")  
        }  
    }  
      
    // Start consuming requests  
    channel.basicConsume(rabbitMQProperties.requestQueue, true, deliverCallback) { consumerTag ->  
        log.info("Consumer was cancelled: $consumerTag")  
    }  
      
    running = true  
    log.info("RabbitMQ server started")  
}

Processing Flow Explained:

• Direct Queue Consumption: Listens directly on request queue (no polling)
• Property Extraction: Safely extracts correlation ID and replyTo from message properties
• Business Logic Processing: Processes the actual request content
• Direct Response Routing: Uses replyTo to send response to correct client instance

Native RPC Response Routing

private fun sendResponse(replyTo: String, correlationId: String, response: String) {  
    // Create response properties preserving correlation ID  
    val props: AMQP.BasicProperties = AMQP.BasicProperties.Builder()  
        .correlationId(correlationId)  
        .build()  
      
    // Send response directly to client's response queue  
    channel.basicPublish("", replyTo, props, response.toByteArray())  
    log.info("Sent response to queue: $replyTo with correlationId: $correlationId")  
}  
  
private fun processBusinessLogic(request: String, correlationId: String): String {  
    // Simulate business processing  
    return "Processed: $request. Correlation ID: $correlationId. ${Random.nextInt(until = 1000)}"  
}

• Direct Routing: No topic/exchange complexity, direct queue-to-queue communication
• Correlation Preservation: Correlation ID maintained in message properties
• Automatic Acknowledgment: Uses auto-acknowledge for simplicity
• Error Isolation: Failed messages don’t affect other processing

Configuration

Application Properties (application.yml)

rabbitmq:  
  host: localhost  
  port: 5672  
  username: anil  
  password: senocak  
  request-queue: request-queue  
  response-queue-prefix: response-queue  
  request-timeout: 30000

Docker Compose Setup

rabbitmq:  
  image: rabbitmq:3.8-management  
  environment:  
    RABBITMQ_DEFAULT_USER: anil  
    RABBITMQ_DEFAULT_PASS: "senocak"  
  ports:  
    - "5672:5672"    # AMQP protocol  
    - "15672:15672"  # Management UI

Advanced Features

Instance Isolation

Each application instance gets its own response queue:

• response-queue-{instanceId} prevents response mix-ups in multi-instance deployments
• Enables horizontal scaling without message conflicts
• Each client instance consumes only its own responses

Native RPC Pattern

• Built-in Correlation: Uses RabbitMQ’s native correlationId property
• Direct Reply Routing: Leverages replyTo property for response routing
• No Custom Headers: Uses standard AMQP message properties
• Simplified Implementation: Less code compared to custom correlation logic

Connection Management

• Automatic Recovery: RabbitMQ client handles connection failures
• Channel Per Component: Separate channels for client and server
• Graceful Shutdown: Proper cleanup of resources on application stop

RabbitMQ vs Kafka Comparison

+---------------------------+--------------------------+------------------------------+----------+
| Feature | RabbitMQ Sync | Kafka Sync | Winner |
+---------------------------+--------------------------+------------------------------+----------+
| Implementation Complexity | Low (native RPC) | High (custom correlation) | RabbitMQ |
| Lines of Code | ~50 lines | ~100+ lines | RabbitMQ |
| Message Properties | Native AMQP properties | Custom headers | RabbitMQ |
| Response Delivery | Push-based (immediate) | Poll-based (100ms delay) | RabbitMQ |
| Latency | Lower (~1-5ms) | Higher (~10-50ms) | RabbitMQ |
| Setup Complexity | Simple queue declaration | Topic + partition management | RabbitMQ |
| Throughput | Medium (10K req/sec) | Very High (100K+ req/sec) | Kafka |
| Persistence | Queue-based | Log-based | Kafka |
| Horizontal Scaling | Good | Excellent | Kafka |
+---------------------------+--------------------------+------------------------------+----------+

When to Use This Pattern

✅ Perfect Use Cases for RabbitMQ Sync

• Low-Latency RPC: Need sub-10ms response times
• Simple Request-Response: Straightforward RPC patterns
• Medium Throughput: Up to 10,000 requests/second
• Traditional Messaging: Familiar queue-based patterns
• Quick Implementation: Need rapid development
• Enterprise Integration: Integrating with existing RabbitMQ infrastructure

❌ When Kafka Sync Might Be Better

• Ultra-High Throughput: >50,000 requests/second
• Event Streaming: Need event log capabilities
• Complex Routing: Multiple consumers per message
• Long-term Storage: Need message replay capabilities

Conclusion

The RabbitMQ synchronous pattern provides a production-ready solution for scenarios requiring:

• Low-latency request-response communication
• Reliable message delivery with built-in acknowledgments
• Simple implementation with native RPC support
• Enterprise-grade features out of the box

This implementation demonstrates RabbitMQ’s strength in traditional messaging patterns while maintaining the scalability benefits of message queuing. It’s ideal for microservice architectures where you need reliable, fast synchronous communication without the complexity of implementing custom correlation logic.

This implementation demonstrates RabbitMQ’s strength in traditional messaging patterns while maintaining enterprise-grade reliability and performance characteristics ideal for synchronous communication requirements.

Full code example can be found in github