Spring Boot 2 - receive response message for the request message using RabbitMQ (RPC)

In this article, we would like to show you how to create Spring Boot 2 application that sends request messages and receives response messages using RabbitMQ RPC pattern.

In Spring Boot 2 application using RabbitMQ RPC pattern, it is necessary to send message from producer to consumer attaching REPLY_TO and CORRELATION_ID headers, that indicate to consumer which producer by what action sent message.

Spring Boot 2 Applications

In this section, we can find a simplified solution that realizes the request-response pattern.

Hint: the article extends project in this article.

Producer application example

The main idea of the producer is to send messages using convertSendAndReceive() method.

ProducerController.java file:

package com.example.demo;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/test")
    @ResponseBody
    public void test() {
        String requestMessage = "Testing request message...";
        String responseMessage = (String) this.rabbitTemplate.convertSendAndReceive("my-queue", requestMessage);
        System.out.println("Response message: " + responseMessage);
    }
}

Hint: if the time of the waiting for the response is long we can change timeout.

Consumer application example

The main idea of the consumer is:

• listen for a request messages,
• send a response message to the queue indicated in AmqpHeaders.REPLY_TO header that indicates the request message producer,
• indicate proper correlation id parameter in the response message to let the producer know on what request message reply is sent.

ConsumerService.java file:

package com.example.demo;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

@Service
public class ConsumerService {

    @Autowired
    private RpcService rpcService;

    @RabbitListener(queues = "my-queue")
    public void handleMyQueue(
            @Header(value = AmqpHeaders.REPLY_TO, required = false) String senderId,
            @Header(value = AmqpHeaders.CORRELATION_ID, required = false) String correlationId,
            String requestMessage
    ) {
        System.out.println("Request message: " + requestMessage);
        if (senderId != null && correlationId != null) {
            String responseMessage = "Testing response message...";
            this.rpcService.sendResponse(senderId, correlationId, responseMessage);
        }
    }
}

RpcService.java file:

package com.example.demo;

import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class RpcService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendResponse(String senderId, String correlationId, Object data) {
        this.rabbitTemplate.convertAndSend(senderId, data, message -> {
            MessageProperties properties = message.getMessageProperties();
            properties.setCorrelationId(correlationId);
            return message;
        });
    }
}

References

1. Direct Reply-to - RabbitMQ Docs

2. Remote procedure call (RPC) - RabbitMQ Docs