Sending and receiving JSON messages with Spring Boot AMQP and RabbitMQ

Setting up a Spring Boot application using AMQP with RabbitMQ is not really difficult, and there are many guides on the Internet showing you the basic setup. But creating an application making use of @RabbitListener annotations and producing and consuming messages in JSON format is trickier, so I would like to share with you a really simple but more serious approach that those hello-messaging apps.

This article covers:

How to send/publish Java Objects as JSON messages using Spring Boot and RabbitMQ RabbitTemplate.
How to read/consume JSON messages to Java Objects using Spring Boot and RabbitMQ @RabbitListener annotation.
How to send and receive Java Objects through RabbitMQ using default Java serializer.

Introduction

As you probably know if you reached this article AMQP is a specification, a protocol that defines messaging and patterns for communication between systems. RabbitMQ is a software that implements AMQP. You can find a lot of tutorials about these concepts on the Internet, apart from many useful books.

Goal

Within this article we’ll set up a Spring Boot application that uses RabbitMQ to send messages to a topic exchange formatted as JSON, and read them using two different queues:

The first Queue will read the contents as a generic org.springframework.amqp.core.Message class.
The second Queue will transform the JSON message to a specific class created by us.

Queues and process view I’ll also show you an alternative configuration to make this work without formatting the message to JSON, but nowadays this format is a good choice for intercommunication between applications and systems that may differ in programming languages, for instance.

I recommend you to check the implementation as you follow these instructions. You can do so by downloading my GitHub project. Don’t forget to give it a star if you find it useful!

Setting up the project

pom.xml

Creating a simple Spring Boot application is pretty straightforward, in my case I use Spring Boot 1.4.1 and Maven. The only thing you need to do to make use of AMQP with RabbitMQ in Spring Boot is including the corresponding library:

<dependency>
   <groupId>org.springframework.amqp</groupId>
   <artifactId>spring-rabbit</artifactId>
   <version>1.6.3.RELEASE</version>
</dependency>

<groupId>org.springframework.amqp</groupId>

<artifactId>spring-rabbit</artifactId>

<version>1.6.3.RELEASE</version>

</dependency>

The message

In this case, we’re going to send a really simple content, but enough to have an idea about how the message looks like in different formats. The class is called CustomMessage.

public final class CustomMessage implements Serializable{

    private String text;
    private int priority;
    private boolean secret;

    // Default constructor is needed to deserialize JSON
    public CustomMessage() {
    }

    public CustomMessage(String text, int priority, boolean secret) {
        this.text = text;
        this.priority = priority;
        this.secret = secret;
    }

    public String getText() {
        return text;
    }

    public int getPriority() {
        return priority;
    }

    public boolean isSecret() {
        return secret;
    }

    @Override
    public String toString() {
        return "CustomMessage{" +
                "text='" + text + '\'' +
                ", priority=" + priority +
                ", secret=" + secret +
                '}';
    }
}

public final class CustomMessage implements Serializable{

private String text;

private int priority;

private boolean secret;

// Default constructor is needed to deserialize JSON

public CustomMessage() {

}

public CustomMessage(String text, int priority, boolean secret) {

this.text = text;

this.priority = priority;

this.secret = secret;

}

public String getText() {

return text;

}

public int getPriority() {

return priority;

}

public boolean isSecret() {

return secret;

}

@Override

public String toString() {

return "CustomMessage{" +

"text='" + text + '\'' +

", priority=" + priority +

", secret=" + secret +

'}';

}

Pretty basic POJO. The only things to point out are the default constructor, which we need for JSON deserialization, and the toString() method, that will give us a human readable representation of the objects.

Sending messages

As mentioned before this example will use a topic exchange to send CustomMessage objects.

@Service
public class CustomMessageSender {

    private static final Logger log = LoggerFactory.getLogger(CustomMessageSender.class);

    private final RabbitTemplate rabbitTemplate;

    @Autowired
    public CustomMessageSender(final RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @Scheduled(fixedDelay = 3000L)
    public void sendMessage() {
        final CustomMessage message = new CustomMessage("Hello there!", new Random().nextInt(50), false);
        log.info("Sending message...");
        rabbitTemplate.convertAndSend(MessagingApplication.EXCHANGE_NAME, MessagingApplication.ROUTING_KEY, message);
    }
}

@Service

public class CustomMessageSender {

private static final Logger log = LoggerFactory.getLogger(CustomMessageSender.class);

private final RabbitTemplate rabbitTemplate;

@Autowired

public CustomMessageSender(final RabbitTemplate rabbitTemplate) {

this.rabbitTemplate = rabbitTemplate;

}

@Scheduled(fixedDelay = 3000L)

public void sendMessage() {

final CustomMessage message = new CustomMessage("Hello there!", new Random().nextInt(50), false);

log.info("Sending message...");

rabbitTemplate.convertAndSend(MessagingApplication.EXCHANGE_NAME, MessagingApplication.ROUTING_KEY, message);

}

Some comments here:

@Scheduled annotation is used to schedule a method to be executed at a given time, or with a given frequency. In this case, this method will be executed every 3 seconds (until the end of the times or you stop the application, whatever happens earlier). You can check more info here.
RabbitTemplate is used to convert and send a message using RabbitMQ. Is a helper class, as many other Template classes existing in Spring (such as JdbcTemplate , RestTemplate , etc.). Spring Boot creates a predefined one for you, but in this case, we need to tune it a little bit to make it use JSON converter when producing messages.
The exchange and the routing key are AMQP concepts that I recommend you to read about that if you still don’t know what they do. As a summary, we are sending messages through a given channel (exchange), and the routing key allows us to filter for who are those generated. This sample application is not intended for showing advanced exchange / queues configuration so you will see that this is a basic setup.

Receiving messages

Of course, the other part of the communication. For this post, I’ve created the receivers in the same application for better logistics (having a single repo), but you won’t have problems to replicate this configuration with two different apps if you want. Actually, you could run this Spring Boot application twice and remove the producer in one instance, the consumers in another one, to make it more like a ‘real case’.

@Service
public class CustomMessageListener {

    private static final Logger log = LoggerFactory.getLogger(CustomMessageListener.class);

    @RabbitListener(queues = MessagingApplication.QUEUE_GENERIC_NAME)
    public void receiveMessage(final Message message) {
        log.info("Received message as generic: {}", message.toString());
    }

    @RabbitListener(queues = MessagingApplication.QUEUE_SPECIFIC_NAME)
    public void receiveMessage(final CustomMessage customMessage) {
        log.info("Received message as specific class: {}", customMessage.toString());
    }
}

@Service

public class CustomMessageListener {

private static final Logger log = LoggerFactory.getLogger(CustomMessageListener.class);

@RabbitListener(queues = MessagingApplication.QUEUE_GENERIC_NAME)

public void receiveMessage(final Message message) {

log.info("Received message as generic: {}", message.toString());

}

@RabbitListener(queues = MessagingApplication.QUEUE_SPECIFIC_NAME)

public void receiveMessage(final CustomMessage customMessage) {

log.info("Received message as specific class: {}", customMessage.toString());

}

Here you can see our goal: the idea is to show at the same time how the same message, being sent by the CustomMessageSender class, can be received through two different queues. First queue (GENERIC) is receiving the message as a generic Message class, so there is no implicit conversion. Spoiler alert: when we log the message we should see something like a JSON string in the payload since it has not been converted.

On the other hand, the queue SPECIFIC is ‘asking for’ a CustomMessage class. That will trigger a logic inside Spring to find a converter from JSON to that specific class. This part should be easy to configure, but sometimes is the part in which we the developers struggle most. In this case is easy to guess that the text logged should be the CustomMessage’s implementation of toString().

Configuration

This is the most interesting part since it’s what is making everything work together. First, you will find some annotations in the class:

@SpringBootApplication
@EnableRabbit
@EnableScheduling
public class MessagingApplication implements RabbitListenerConfigurer {

@SpringBootApplication

@EnableRabbit

@EnableScheduling

public class MessagingApplication implements RabbitListenerConfigurer {

Apart from the one that makes Spring Boot works, @EnableRabbit is making sure that Rabbit Listeners and other Rabbit configurations are detected and processed. @EnableScheduling needs to be added to support the @Scheduled method for sending messages that we covered before in CustomMessageSender class.

Then you’ll find some constants and, next to them, the Exchange / Queues configuration:

@Bean
public TopicExchange appExchange() {
    return new TopicExchange(EXCHANGE_NAME);
}

@Bean
public Queue appQueueGeneric() {
    return new Queue(QUEUE_GENERIC_NAME);
}

@Bean
public Queue appQueueSpecific() {
    return new Queue(QUEUE_SPECIFIC_NAME);
}

@Bean
public Binding declareBindingGeneric() {
    return BindingBuilder.bind(appQueueGeneric()).to(appExchange()).with(ROUTING_KEY);
}

@Bean
public Binding declareBindingSpecific() {
    return BindingBuilder.bind(appQueueSpecific()).to(appExchange()).with(ROUTING_KEY);
}

@Bean

public TopicExchange appExchange() {

return new TopicExchange(EXCHANGE_NAME);

}

@Bean

public Queue appQueueGeneric() {

return new Queue(QUEUE_GENERIC_NAME);

}

@Bean

public Queue appQueueSpecific() {

return new Queue(QUEUE_SPECIFIC_NAME);

}

@Bean

public Binding declareBindingGeneric() {

return BindingBuilder.bind(appQueueGeneric()).to(appExchange()).with(ROUTING_KEY);

}

@Bean

public Binding declareBindingSpecific() {

return BindingBuilder.bind(appQueueSpecific()).to(appExchange()).with(ROUTING_KEY);

}

These are the topic exchange, both queues and the bindings between them (both queues are bound to the same exchange with the same routing key). Just a sample configuration.

Below in the same class, we can find the configuration related to producing messages:

@Bean
public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
    final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());
    return rabbitTemplate;
}

@Bean
public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
    return new Jackson2JsonMessageConverter();
}

@Bean

public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {

final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());

return rabbitTemplate;

}

@Bean

public Jackson2JsonMessageConverter producerJackson2MessageConverter() {

return new Jackson2JsonMessageConverter();

}

The only thing we’re doing here is overriding the default RabbitTemplate created by Spring Boot with another one that will use a message converter of type Jackson2JsonMessageConverter. Remember: this configuration only affects to messages being sent by RabbitTemplate.

And then we need to configure how our application should interpret the received messages. You need to implement a method from the interface RabbitListenerConfigurer, responsible for handling the RabbitListener annotations. As you can see below, we need to configure the MessageHandlerMethodFactory to be not plainly the default one, but using a JSON message converter. Take into account that this converter is not the same as configured for producing messages since we’re using different classes.

@Bean
public MappingJackson2MessageConverter consumerJackson2MessageConverter() {
   return new MappingJackson2MessageConverter();
}

@Bean
public DefaultMessageHandlerMethodFactory messageHandlerMethodFactory() {
   DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
   factory.setMessageConverter(consumerJackson2MessageConverter());
   return factory;
}

@Override
public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {
   registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
}

@Bean

public MappingJackson2MessageConverter consumerJackson2MessageConverter() {

return new MappingJackson2MessageConverter();

}

@Bean

public DefaultMessageHandlerMethodFactory messageHandlerMethodFactory() {

DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();

factory.setMessageConverter(consumerJackson2MessageConverter());

return factory;

}

@Override

public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {

registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());

}

Running the application

First, you need to make sure is that RabbitMQ server is running. Also, if your configuration is different from the default one, you need to set it up correctly within the application.properties file. After executing the app you should see these lines in the console:

es.macero.CustomMessageSender            : Sending message...
es.macero.CustomMessageListener          : Received message as specific class: CustomMessage{text='Hello there!', priority=3, secret=false}
es.macero.CustomMessageListener          : Received message as generic: (Body:'{"text":"Hello there!","priority":3,"secret":false}' MessageProperties [headers={__TypeId__=es.macero.CustomMessage}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/json, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=appExchange, receivedRoutingKey=messages.key, receivedDelay=null, deliveryTag=2, messageCount=0, consumerTag=amq.ctag-N2Gh7LkIyDAXkPDzd2Gm0A, consumerQueue=appGenericQueue])

es.macero.CustomMessageSender : Sending message...

es.macero.CustomMessageListener : Received message as specific class: CustomMessage{text='Hello there!', priority=3, secret=false}

es.macero.CustomMessageListener : Received message as generic: (Body:'{"text":"Hello there!","priority":3,"secret":false}' MessageProperties [headers={__TypeId__=es.macero.CustomMessage}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/json, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=appExchange, receivedRoutingKey=messages.key, receivedDelay=null, deliveryTag=2, messageCount=0, consumerTag=amq.ctag-N2Gh7LkIyDAXkPDzd2Gm0A, consumerQueue=appGenericQueue])

Working as expected! Our listener for CustomMessage is showing us our implementation of toString() method. The last line though, the method listening for the Message class, is showing us the raw contents, which turns out to be –of course– JSON format.

Changing from JSON format to default

I invite you to comment out the lines for setting up the producer and consumer message converters (they are indicated in the source code). Then, remove the implements clause from the configuration class. What you will see then is the default configuration for Spring Boot, which is using a Java-specific format to encode messages.

es.macero.CustomMessageSender            : Sending message...
es.macero.CustomMessageListener          : Received message as specific class: CustomMessage{text='Hello there!', priority=7, secret=false}
es.macero.CustomMessageListener          : Received message as generic: (Body:'CustomMessage{text='Hello there!', priority=7, secret=false}' MessageProperties [headers={}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/x-java-serialized-object, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=appExchange, receivedRoutingKey=messages.key, receivedDelay=null, deliveryTag=2, messageCount=0, consumerTag=amq.ctag-Aobj21v_HVDXwAjRn2i3_Q, consumerQueue=appGenericQueue])

es.macero.CustomMessageSender : Sending message...

es.macero.CustomMessageListener : Received message as specific class: CustomMessage{text='Hello there!', priority=7, secret=false}

es.macero.CustomMessageListener : Received message as generic: (Body:'CustomMessage{text='Hello there!', priority=7, secret=false}' MessageProperties [headers={}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/x-java-serialized-object, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=appExchange, receivedRoutingKey=messages.key, receivedDelay=null, deliveryTag=2, messageCount=0, consumerTag=amq.ctag-Aobj21v_HVDXwAjRn2i3_Q, consumerQueue=appGenericQueue])

As you can see in the headers of the message now the format is contentType=application/x-java-serialized-object. This is just another approach, which could turn out to be not the best one if you are using, for instance, different kind of applications (not-Java) for receiving messages.

Thanks and… do you want to know more?

That’s all, I hope you find this how-to guide useful. As mentioned before you can clone the repository with the code ready to run from my GitHub account:

https://github.com/mechero/spring-boot-amqp-messaging (give me a star if you think is useful!)

Also, if you are interested in how to apply Event Driven patterns in a Microservice Architecture and some other related topics like Service Discovery, Routing, etc., you can get my book Microservices – The Practical Way, which will guide you step by step to get real experience.

Hystrix fallback with Zuul and Spring Boot When we work with a Gateway Service like Zuul, probably we want to include a Circuit Breaker mechanism to avoid ugly errors in case of redirecting to ...

Mapping REST API to Plain Java Objects (POJOs) Let's say you want to connect your web/application with a third party REST API. Within this article I'll put as example the SonarQube API, given that ...


February 22, 2017
paris

Thanks for this article, I’m relatively new to spring and I found it very useful. Good luck with your book 🙂
- Reply
  
  February 22, 2017
  Moisés
  
  Thanks for your feedback! I’ll try to cover this and some other topics on my book so feel free to show your interest on Leanpub if you want to get a good discount!

February 25, 2017
Naveen Kumar K

Hi MOISÉS, You saved my day! Wonderful post…

April 7, 2017
Maneesh Mohanan

Thank you very much for the article ! It was really helpful.

May 25, 2017
Peter

Nice post! It helps us a lot with migration of our rabbits to spring boot

July 3, 2017
aditya prakash Bv

Thank you ver much cheers!!!!

July 27, 2017
Anonymous

Thanks, can u do Event Sourcing with RabbitMq
- Reply
  
  July 27, 2017
  Moisés
  
  Technically you can, but you still need the event store so it’s better to go for a framework that can provide you all you need, you’ll get a better integration. Have a look at https://www.confluent.io/blog/event-sourcing-cqrs-stream-processing-apache-kafka-whats-connection/, about how to do Event Sourcing with Kafka.