Java
Spring Boot and RabbitMQ. Part 1

Spring Boot and RabbitMQ. Part 1

After long break I’m back with some new interesting topic for you. I decided to bring my blog back to life with regular posting. Hopefully time allows me to do it more often than one time a year 🙂

In my new small serie of two blog posts I will introduce you RabbitMQ, one of two most popular message brokers. Second one is Apache Kafka which is more powerful, but if you need only messaging functionality choosing it can be overkill, because in most of cases it requires more expertise. RabbitMQ can also handle high throughput and offers good reliability by delivering of messages. Setting up is quite simple, so we could have in short time ready to go messaging setup in our distributed system.

In this post I’ll make quick introduction into RabbitMQ with it’s basics concepts, like exchange, queue, binding, routing key. Described topics I’ll explain with sample code. Whole project can be found on my GitHub – https://github.com/szymon-sawicki/rabbitmq-products-app.

In second blog post I’ll show you how to create dead letter queue, make some integration tests with testcontainers and introduce automatic queues/exchanges configuration by the startup of docker container.

List of contents

  1. Short introduction to RabbitMQ
  2. Example project
    a. Producer
    b. Consumer
  3. Exchanges
    a. Direct exchange
    b. Fanout exchange
    c. Topic exchange
    d. Header exchange
  4. Summary

Short introduction to RabbitMQ

RabbitMQ uses AMQP (Advanced Message Queuing Protocol) protocol which is defining message oriented semantics, is technology independent and easy to implement. With additional plugins other protocols could be also used here.

Basic concepts:

Exchange – routes messages sent from producer to queue (or queues).
Queue – stores the messages forwarded from exchange, using FIFO data structure (first in, first out)
Binding – links queue and exchange
Routing key – used by exchange to find out, which to which queue message should be sent

Example project

Project contains two services – products producer, products service and RabbitMQ broker:

This is how packages and files are structured, here is nothing extraordinary. I not used any special design approach:

Services are running on Java 21 with Spring Boot 3.2. Whole application can be started with docker compose (https://github.com/szymon-sawicki/rabbitmq-products-app/blob/main/docker-compose.yml).

Spring Boot offers nice starter containing all necessary components for RabbitMQ, it reduces overhead and simplifies development process. W need only introduce this dependency (in build.gradle):

Setup of exchanges and queues

After starting of containers you can log in into admin GUI (http://localhost:15672/, login: guest, password: guest). In this case we create topics and exchanges here, in normal commercial usage it will be done probably by admin/devops. Spring Boot AMQP offers this possiblity also using annotations and beans, but I won’t describe it here – it not necessary, because in most of cases and you’ll not need it.

All queues and exchanges which needs to be created are described in readme file of the project. It’s important to make it before starting of other services, because the product service will throw some exceptions and won’t start.

There is also possibilty to create all this configuration on the startup of container, I will introduce this soulution in the second part of this blog post series.

Producer

Producer is responsible for sending of messages to broker. In our case I will expose one REST endpoint accepting type of exchange as path variable and product as body. Format of the body:

Controller (ProductsController) takes the request and pass it to service (ProductProducer). Service switch statement resolves type of the exchange and introduces specific logic.

Product object is serialized to JSON format with ObjectMapper before sending the to queue. For the routing exchanges of type direct and topic are using routing key, which we are generating with this method:

In case of exchange type “header” routing key is ignored, but we need some extra logic to include headers in the message. To achieve this, new object of type MessageProperties is created – here we are defining the custom headers. Then new instance of SimpleMessageConverter is created, aferwards we are sending new message with JSON body and MessageProperties with toMessage method:

If the message is created, it can be send to the queue using RabbitTemplate (it is part of AMQP Spring Boot Starter) and convertAndSend method:

Consumer (ProductsService)

On the other side we can find consumer, which is recieving messages from RabbitMQ broker. In sample project I created separate consumer class for each type of exchange and one endpoint printing all recieved products (stored in memory in form of list).

To consume messages from queue we need to introduce @RabbitListener annotation on method level with name of the queue as argument, this queue must be created before, otherwise it’ll not work. This is example from ProductConsumerTopic class:

Exchanges

Exchange is responsible for the routing of messages to proper queue. It’s the first place in broker where messages from producer are landing. Decision where the message should be routed is based on type of exchange and routing key, or headers.

Direct exchange:

This type of exchange redirects messages to queue based on the routing key. It can be one (unicast) or more queues (multicast). In our sample project we created bindings representing castegories of products (books, clothings, electronics) that are basis for redirecting messages to proper queues (q.books, q.clothing, q.electronics). The category of product is used as routing key.

RabbitMQ direct exchange

Fanout exchange:

Kind of exchange which is good for broadcasting, in this case messages are sent to all bound queues and the binding key is ignored. Here I created two queues (q.warehouse, q.archive) and bound it to exchange.

RabbitMQ fanout exchange


Topic exchange:

This type of exchange is similar to direct echange and redirection is based on routing key. But in this case we can add some extra logic. Routing key must here be costructed with some elements splited by “.”, then we can use wildcard to match the routing. We could choose from two wildcards hash (“#”) which is matching zero or more words and star (“*”) which is matching only one word.
In our project by topic exchange I will construct routing key with category + resolved price range (less than 100 is ‘low’, between 100 and 200 is ‘medium’ and everything above is high). Then in exchange I created 3 bindings with routing keys:

  • BOOKS.# that routes all products in books catgory, independent of range to queue q.books
  • #.low routes all products with price lower than 100 to q.cheap-products queue
  • CLOTHING.high sending all clothing with price higher than 200 to q.expensive-clothings queue



Header exchange:

The Header Exchange allows for more complex routing scenarios where routing is based on message header values instead of the routing key. Unlike Direct and Topic exchanges, the Header exchange ignores the routing key. Instead, it uses message headers for routing decisions. This is particularly useful for scenarios where the message’s context or content, represented by headers, dictates its routing.

Matching Strategies

With header exchanges, you can specify whether a message’s headers need to match all the header values specified in the binding (all match) or just one (any match). This is determined by adding a special argument (x-match) to the binding:

x-match: all — This is the default behavior. All the header key-value pairs specified in the binding must match those present in the message for it to be routed to the bound queue.

x-match: any — Only one of the header key-value pairs needs to match for the message to be routed to the bound queue.

To show how it works, I created two bindings, each with three arguments. First one (x-match: any) redirects messages with category BOOKS, or CLOTHINGS to queue q.books-clothings. Second binding (x-match: all) sends products with category CLOTHING and price range “high” to q.expensive-clothings queue.

In producer I’m attaching additional headers (I derscribed it earlier in this post), which are used by exchange for routing.

Summary

To wrap up, RabbitMQ offers a flexible and powerful platform for handling messaging in distributed systems. Its support for multiple exchange types allows developers to implement a wide range of messaging patterns, from simple point-to-point communication to complex topic-based routing scenarios. The Spring Boot AMQP starter simplifies the integration of RabbitMQ into your Java applications, providing annotations and configuration options to manage connections, message listeners, and message converters.

As we explored through practical examples, whether you’re broadcasting messages to multiple consumers, routing based on specific criteria, or even customizing message delivery based on headers, RabbitMQ coupled with Spring Boot offers a robust solution.

Remember, the example code provided in this post and the complete project on GitHub are starting points. Experiment with them, tweak the configurations, and explore RabbitMQ’s features to fit your application’s needs. Happy coding, and stay tuned for more posts on advanced RabbitMQ topics and other interesting technologies.