Messaging with RabbitMQ and Spring Boot

RabbitMQ is an open-source message broker that implements several messaging protocols, including AMQP and STOMP. It is frequently used to queue messages and interact with queues. A RabbitMQ server acts as an exchange server (or broker). Clients, written in a variety of languages, enable applications to publish and consume the messages in the queue.

In this post, we'll learn how to create a queue with RabbitMQ and interact with it using Spring Boot.

Create a Spring Boot application with spring-boot-starter-amqp and jackson-databind as the dependencies.

Create a RabbitMQ broker

Let's start by setting up a RabbitMQ broker; you can choose to install it on your machine or run it as a container. For later, create a docker-compose.yml file in the project's root and add the following configuration.

 1version: '3.1'
 2
 3services:
 4
 5  rabbitmq:
 6    image: "rabbitmq:3-management"
 7    container_name: "rmq3"
 8    hostname: "albatross"
 9    restart: "no"
10    environment:
11      RABBITMQ_DEFAULT_USER: "rabbitmq"
12      RABBITMQ_DEFAULT_PASS: "rabbitmq"
13    labels:
14      NAME: "rabbitmq1"
15    ports:
16      - "5672:5672"
17      - "15672:15672"

Now, execute the following command on CLI.

docker-compose up -d

This will launch a RabbitMQ3 container. To access the management console, point your browser at http://localhost:15672/ and login with the same username and password as set in docker-compose.yml.

Define a domain

Say, you want to publish a list of books in a message and then consume it. To do so, define a simple Book class as follows.

 1// src/main/java/dev/mflash/guides/rabbitmq/domain/Book.java
 2
 3public class Book {
 4
 5  private final String isbn;
 6  private final String title;
 7  private final String author;
 8
 9  // constructors, getters and setters, etc.
10}

Let's create a list of books to publish them on a queue.

Configure Queue, Topic Exchange and Routing Key

A typical RabbitMQ queue has

• a name to identify it,
• an optional routing key to selectively process messages, and
• an exchange to route the messages to a queue based on the value of a routing key.

A topic exchange works on a wildcard match of a routing pattern. Topic exchange is not the only type of exchange available for use but it'll suffice for this guide. To bind a queue with an exchange and a routing key, we'll have to create a Binding which can be injected through a Bean as follows.

 1// src/main/java/dev/mflash/guides/rabbitmq/configuration/RabbitMQConfiguration.java
 2
 3public @Configuration class RabbitMQConfiguration {
 4
 5  public static final String TOPIC_EXCHANGE_NAME = "mflash-exchange";
 6  public static final String QUEUE_NAME = "mflash-queue";
 7  private static final String ROUTING_KEY = "books.#";
 8
 9  public @Bean TopicExchange topicExchange() {
10    return new TopicExchange(TOPIC_EXCHANGE_NAME);
11  }
12
13  public @Bean Queue queue() {
14    return new Queue(QUEUE_NAME);
15  }
16
17  public @Bean Binding binding() {
18    return BindingBuilder.bind(queue()).to(topicExchange()).with(ROUTING_KEY);
19  }
20}

Create a Publisher

A publisher or producer sends the messages in a queue. In our case, it’ll send a list of books. We'll use a RabbitTemplate object injected through Spring to send this list in the queue.

 1// src/main/java/dev/mflash/guides/rabbitmq/service/Publisher.java
 2
 3public @Service class Publisher implements CommandLineRunner {
 4
 5  private final RabbitTemplate rabbitTemplate;
 6  private final Reader reader;
 7
 8  public Publisher(RabbitTemplate rabbitTemplate, Reader reader) {
 9    this.rabbitTemplate = rabbitTemplate;
10    this.reader = reader;
11  }
12
13  public @Override void run(String... args) throws Exception {
14    rabbitTemplate
15        .convertAndSend(RabbitMQConfiguration.TOPIC_EXCHANGE_NAME, "books.new", getBooks());
16    reader.getLatch().await(10, TimeUnit.SECONDS);
17  }
18
19  private List<Book> getBooks() {
20    List<Book> books = new ArrayList<>();
21    books.add(new BookBuilder().isbn("978-1250078308").title("Archenemies").author("Marissa Meyer")
22        .build());
23    books.add(new BookBuilder().isbn("978-0399555770").title("Skyward").author("Brandon Sanderson")
24        .build());
25    books.add(new BookBuilder().isbn("978-0374285067").title("Void Star").author("Zachary Mason")
26        .build());
27
28    return books;
29  }
30}

Note that all the published messages are serialized as byte arrays by default. To properly serialize the list of Books, we need to define a message converter for RabbitTemplate as an instance of Jackson2JsonMessageConverter.

 1// src/main/java/dev/mflash/guides/rabbitmq/configuration/RabbitMQConfiguration.java
 2
 3public @Configuration class RabbitMQConfiguration {
 4
 5  public static final String TOPIC_EXCHANGE_NAME = "mflash-exchange";
 6  public static final String QUEUE_NAME = "mflash-queue";
 7  private static final String ROUTING_KEY = "books.#";
 8
 9  public @Bean TopicExchange topicExchange() {
10    return new TopicExchange(TOPIC_EXCHANGE_NAME);
11  }
12
13  public @Bean Queue queue() {
14    return new Queue(QUEUE_NAME);
15  }
16
17  public @Bean Binding binding() {
18    return BindingBuilder.bind(queue()).to(topicExchange()).with(ROUTING_KEY);
19  }
20
21  public @Bean RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
22    final var rabbitTemplate = new RabbitTemplate(connectionFactory);
23    rabbitTemplate.setMessageConverter(messageConverter());
24    return rabbitTemplate;
25  }
26
27  public @Bean MessageConverter messageConverter() {
28    return new Jackson2JsonMessageConverter();
29  }
30}

Create a Consumer

A reader or consumer will read the messages published by the publisher. In our case, we'll simply print the list.

 1// src/main/java/dev/mflash/guides/rabbitmq/service/Reader.java
 2
 3public @Service class Reader {
 4
 5  private CountDownLatch latch = new CountDownLatch(1);
 6
 7  @RabbitListener(queues = RabbitMQConfiguration.QUEUE_NAME, containerFactory = "listenerContainerFactory")
 8  public void receiveMessage(final List<Book> books) {
 9    books.forEach(System.out::println);
10    latch.countDown();
11  }
12
13  public CountDownLatch getLatch() {
14    return latch;
15  }
16}

A CountDownLatch is used to wait for several threads to complete (here, it is set to 1).

To convert the incoming message into a list of books, we'll have to provide the MessageConverter to a RabbitListener. This can be done by injecting the MessageConverter through an instance of SimpleRabbitListenerContainerFactory as follows.

 1// src/main/java/dev/mflash/guides/rabbitmq/configuration/RabbitMQConfiguration.java
 2
 3public @Configuration class RabbitMQConfiguration {
 4
 5  public static final String TOPIC_EXCHANGE_NAME = "mflash-exchange";
 6  public static final String QUEUE_NAME = "mflash-queue";
 7  private static final String ROUTING_KEY = "books.#";
 8
 9  public @Bean TopicExchange topicExchange() {
10    return new TopicExchange(TOPIC_EXCHANGE_NAME);
11  }
12
13  public @Bean Queue queue() {
14    return new Queue(QUEUE_NAME);
15  }
16
17  public @Bean Binding binding() {
18    return BindingBuilder.bind(queue()).to(topicExchange()).with(ROUTING_KEY);
19  }
20
21  public @Bean SimpleRabbitListenerContainerFactory listenerContainerFactory(
22      ConnectionFactory connectionFactory) {
23    final var factory = new SimpleRabbitListenerContainerFactory();
24    factory.setConnectionFactory(connectionFactory);
25    factory.setMessageConverter(messageConverter());
26    return factory;
27  }
28
29  public @Bean RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
30    final var rabbitTemplate = new RabbitTemplate(connectionFactory);
31    rabbitTemplate.setMessageConverter(messageConverter());
32    return rabbitTemplate;
33  }
34
35  public @Bean MessageConverter messageConverter() {
36    return new Jackson2JsonMessageConverter();
37  }
38}

When the application is launched, the publisher will publish the list of books on a queue called mflash-queue. After 10 seconds, the consumer will be called to print the message received from the queue.

References

Source Code — spring-messaging-rabbitmq