RabbitMQ - Pub-Sub Pattern: Implementation Of Fanout Exchange

In this article, we will learn about the messaging patterns of message brokers. RabbitMQ supports a wide range of messaging patterns one of which is the Publisher-Subscriber (Pub-Sub) pattern. In the previous article, we learned what a message broker is.

Pub/Sub Pattern

Assume I've published an article on Medium. One Medium user liked my article and gave me a clap👏. Now Medium would do the following: increase the clap count for that article by one, notify me via email that one user had clapped my article, and pay me 10 cents(!).

So, based on a single event, different services are triggered. In the same way, in the Pub-Sub pattern, a producer sends a message to an exchange, which then sends the same message to multiple consumers. The reason for doing so is that different services in the microservice architecture may be interested in processing the same message. Similar to the preceding example, the various services process the clapping information differently.

Consider another example. When a user creates a new account in an application, the new user information is stored in one microservice. This data may be of interest to the logging or auditing microservice. And the promotional service would use these user details to send promotional emails.

Exchange is extremely important in the Pub-Sub pattern. In the previous article, the producer is connected to a specific queue via the default exchange. However, in the Pub-Sub pattern, the producer can only send messages to the exchange. The exchange must know what to do with the message that arrives. RabbitMQ provides various exchange types to help you decide what to do with the message. The available exchange types are listed below.

• Direct
• Topic
• Header
• Fanout

The Fanout exchange publishes the same message to multiple queues. We will use a Fanout exchange to implement the Pub-Sub pattern.

Random Queue Allocation

In our previous article, we described how the producer communicates with the consumer via a specified queue. This means whenever the producer wants to exchange messages it must use that particular queue to send messages to consumers.

However, in this pattern, we are only interested in current messages, not old ones. To accomplish this, we could create a queue with a random name or allow the channel to select a random queue name. The queue is automatically deleted once the consumer disconnects from the exchange.

Implementation using Java

Shared Component Module

The Shared Component Module contains the MessageBody class. It consists of two records: UserDetails and ArticleDetails.

package com.raven.components.model;

public record UserDetails(
        String name,
        String email
) {
}

package com.raven.components.model;

public record ArticleDetails(
        String length,
        String publicationDate,
        Boolean isClapped,
        String clappedBy,
        Integer readTime,
        String timeUnit
) {
}

package com.raven.components.model;

public class MessageBody {
    private UserDetails userDetails;
    private ArticleDetails articleDetails;

    public UserDetails getUserDetails() {
        return userDetails;
    }

    public MessageBody setUserDetails(UserDetails userDetails) {
        this.userDetails = userDetails;
        return this;
    }

    public ArticleDetails getArticleDetails() {
        return articleDetails;
    }

    public MessageBody setArticleDetails(ArticleDetails articleDetails) {
        this.articleDetails = articleDetails;
        return this;
    }
}

We will use MessageBody to transport data between producers and consumers.

This module also includes the UtilityService class, which contains a variety of conversion methods.

package com.raven.components.utility;

public class UtilityService {

    public static byte[] convertObjectToByte() {
        var messageBody = getMessage();
        var mapper = new ObjectMapper();
        byte[] _byte = new byte[0];

        try {
            String message = mapper.writeValueAsString(messageBody);
            _byte = message.getBytes(StandardCharsets.UTF_8);
        } catch (JsonProcessingException e) {
            System.out.println("Error in processing the JSON : " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }

        return _byte;
    }

    public static byte[] convertObjectToByte(MessageBody messageBody) {
        var mapper = new ObjectMapper();
        byte[] _byte = new byte[0];

        try {
            String message = mapper.writeValueAsString(messageBody);
            _byte = message.getBytes(StandardCharsets.UTF_8);
        } catch (JsonProcessingException e) {
            System.out.println("Error in processing the JSON : " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }

        return _byte;
    }

    public static String convertObjectToString() {
        var messageBody = getMessage();
        var mapper = new ObjectMapper();
        String message = "";

        try {
            message = mapper.writeValueAsString(messageBody);
        } catch (JsonProcessingException e) {
            System.out.println("Error in processing the JSON : " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }

        return message;
    }

    public static String convertObjectToString(MessageBody messageBody) {
        var mapper = new ObjectMapper();
        String message = "";

        try {
            message = mapper.writeValueAsString(messageBody);
        } catch (JsonProcessingException e) {
            System.out.println("Error in processing the JSON : " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }

        return message;
    }

    public static MessageBody convertStringToObject(String message) {
        var mapper = new ObjectMapper();
        MessageBody messageBody = new MessageBody();

        try {
            messageBody = mapper.readValue(message, new TypeReference<MessageBody>() {
            });
        } catch (JsonProcessingException e) {
            System.out.println("Error in processing the JSON : " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }

        return messageBody;
    }

    public static MessageBody convertStringToObject(byte[] bytes) {
        var mapper = new ObjectMapper();
        MessageBody messageBody = new MessageBody();

        try {
            String message = new String(bytes, StandardCharsets.UTF_8);
            messageBody = mapper.readValue(message, new TypeReference<MessageBody>() {
            });
        } catch (JsonProcessingException e) {
            System.out.println("Error in processing the JSON : " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }

        return messageBody;
    }

    private static MessageBody getMessage() {
        return new MessageBody()
                .setUserDetails(new UserDetails(
                        "Paula Small",
                        "paula.small@bilearner.com")
                )
                .setArticleDetails(new ArticleDetails(
                        "1289 words",
                        "03/12/2024",
                        true,
                        "john.dow@yahoomail.com",
                        80,
                        "second")
                );
    }
}

We will use these conversion methods to send and receive messages between producers and consumers.

Producer

To implement the Pub-Sub pattern, we defined an exchange of type FANOUT. We are publishing the MessageBody object to the exchange. As you can see, we are not attaching a queue to the channel. The channel is directly connected to the exchange.

package com.raven.producer;

public class MyProducer {
    private static final String HOST = "localhost";
    private static final String EXCHANGE = "pub_sub";

    public static void main(String[] args) {
        // CREATE A CONNECTION FACTORY
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(HOST);

        // CREATE A CONNECTION FROM FACTORY
        try (Connection connection = connectionFactory.newConnection()) {
            // GET CHANNEL FROM CONNECTION
            Channel channel = connection.createChannel();

            // DECLARE A EXCHANGE
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);

            // MESSAGE DETAILS
            var message = UtilityService.convertObjectToString(
                    new MessageBody()
                            .setUserDetails(new UserDetails(
                                    "Paula Small",
                                    "paula.small@bilearner.com")
                            )
                            .setArticleDetails(new ArticleDetails(
                                    "1289 words",
                                    "03/12/2024",
                                    true,
                                    "john.dow@yahoomail.com",
                                    80,
                                    "second")
                            )
            );

            // PUBLISH A MESSAGE TO CHANNEL
            channel.basicPublish(EXCHANGE, "", null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" Message sent : '" + message + "'");
            System.out.println();

        } catch (Exception e) {
            System.out.println("Error in sending message: " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }
    }
}

Consumer #1

Here is our first consumer: Article service. Here, we are processing the article's clap information. The queueDeclare() method is used to declare an auto-delete, non-durable queue. It returns the queue name once the queue has been successfully created. The queue is then bound to the exchange with the queueBind() method.

package com.raven.article_cosumer;

public class ArticleConsumer {
    private static final String HOST = "localhost";
    private static final String EXCHANGE = "pub_sub";

    public static void main(String[] args) {
        // CREATE A CONNECTION FACTORY
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(HOST);

        try {
            // CREATE A CONNECTION
            Connection connection = connectionFactory.newConnection();

            // CREATE A CHANNEL FROM CONNECTION
            Channel channel = connection.createChannel();

            // DECLARE A EXCHANGE
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);

            // DECLARE A QUEUE IN THE CHANNEL AND GET ITS NAME
            String queueName = channel.queueDeclare().getQueue();
            System.out.println("Consumer: Article Service: Queue name : " + queueName);

            // BIND THE QUEUE WITH THE EXCHANGE
            channel.queueBind(queueName, EXCHANGE, "");

            System.out.println("Waiting for message...");

            DeliverCallback deliverCallback = (consumeMsg, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println("Consumer: Article Service :: Message received : '" + message + "'");

                System.out.println();
                var messageBody = UtilityService.convertStringToObject(message);

                ArticleDetails articleDetails = messageBody.getArticleDetails();
                System.out.println("-> Article details: " + articleDetails);

                if (articleDetails.isClapped()) {
                    System.out.println("-> Article is clapped by " + articleDetails.clappedBy());
                    System.out.println("-> And article has 67 claps now!");
                }
            };

            // GET MESSAGE FROM EXCHANGE
            channel.basicConsume(queueName, true, deliverCallback, consumeMsg -> {
            });
        } catch (Exception e) {
            System.out.println("Article Consumer: Error in consuming message: " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }
    }
}

Consumer #2

Payment service is our second consumer. This consumer handles the payment information. Here, we create the queue at runtime and associate it with the exchange.

package com.raven.payment_cosumer;

public class PaymentConsumer {
    private static final String HOST = "localhost";
    private static final String EXCHANGE = "pub_sub";

    public static void main(String[] args) {
        // CREATE A CONNECTION FACTORY
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(HOST);

        try {
            // CREATE A CONNECTION
            Connection connection = connectionFactory.newConnection();

            // CREATE A CHANNEL FROM CONNECTION
            Channel channel = connection.createChannel();

            // DECLARE A EXCHANGE
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);

            // DECLARE A QUEUE IN THE CHANNEL AND GET ITS NAME
            String queueName = channel.queueDeclare().getQueue();
            System.out.println("Consumer: Payment Service: Queue name : " + queueName);

            // BIND THE QUEUE WITH THE EXCHANGE
            channel.queueBind(queueName, EXCHANGE, "");

            System.out.println("Waiting for message...");

            DeliverCallback deliverCallback = (consumeMsg, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println("Consumer: Article Service :: Message received : '" + message + "'");

                System.out.println();
                var messageBody = UtilityService.convertStringToObject(message);

                ArticleDetails articleDetails = messageBody.getArticleDetails();
                System.out.println("-> Article details: " + articleDetails);

                if (articleDetails.isClapped() && articleDetails.readTime() > 50) {
                    System.out.println("-> Article is read time is " + articleDetails.readTime() + articleDetails.timeUnit());
                    System.out.println("-> You have received 30 cents.");
                    System.out.println("-> Your total earning is $10.45.");
                }
            };

            // GET MESSAGE FROM EXCHANGE
            channel.basicConsume(queueName, true, deliverCallback, consumeMsg -> {
            });
        } catch (Exception e) {
            System.out.println("Payment Consumer: Error in consuming message: " + e.getMessage() + ", " + Arrays.toString(e.getStackTrace()));
        }
    }
}

Testing

We need to install RabbitMQ on our local machine to communicate with the Java application. You can find an installation guide here. So, we'll start our two consumer applications one at a time, then the producer program.

The producer sends the message.

The article service consumer receives and processes the message.

The payment service consumer also receives and processes the message.

The RabbitMQ management console is accessible via http://localhost:15672/#/. To log in to the console, use guest as the username and password. Click the 'Exchanges' tab to view the exchanges that are linked to the message broker. Find our exchange, 'pub_sub', and click on it for more information.

To view the queues, select the 'Queues and Streams' tab.

Source code: PubSubDemo.

Happy coding!!! 😊

in

RabbitMQ: Message Broker - A Brief Introduction

We are all familiar with Hogwarts' magical postal network, the Owl Post Office. This magical delivery system relies on owls to deliver packages and letters. They can even pick up an item from any address and give it to another using their magical tracking abilities. Similarly, we can compare RabbitMQ, a message broker to the Owl Post Office. It allows applications to interact with one another and exchange messages.

But why should we use message broker?

An e-commerce application ships thousands of items per day and sends email notifications for each one. This is a synchronous operation. This means that the item's ship status is saved in the database, and an email is sent. Now, on bad days, the email server goes down or crashes due to overload. So none of us receive the email notification.

Assume we add a service layer between the shipment service and e-mail service. So, shipping service sends the shipment notification to the service layer, which will route the message to the email service. If the email service goes down, the service layer will store the messages. When the email service goes live, the service layer will push those messages to the e-mail service.

And this service layer is nothing but a message broker.

RabbitMQ is a distributed message and stream broker. A message broker is software that sits among applications, allowing them to exchange messages. RabbitMQ is useful for decoupling services, remote procedure calls (RPC), streaming services, and IoT.

RabitMQ is most commonly used in microservice-based architectures. It operates asynchronously. This means they do not follow a simple request-response pattern, and we must wait for replies. RabbitMQ, like the postal service, sends messages from producer to consumer.

Producer

A Producer is an action or event that generates messages. Credit card transactions, a drop or rise in stock price, or an order dispatch are all examples of Producers.

Consumer

On the other hand, Consumers are the entities that listen to messages. Humans are the perfect example of a consumer. We consume everything, from news to alcohol. Jokes aside! Because a message broker might be connected to multiple producers and consumers, communication between them is asynchronous.

Exchange

But, wait a second! How does the message broker get a message from the producer to the consumer? The answer is Exchange. It functions as RabbitMQ's brain. Exchange helps message broker to route messages from producer to consumer.

Queue & Binding

A message broker may have multiple exchanges. Exchanges always receive messages from producers. Consumers are not connected directly to exchanges. Queues connect exchanges and consumers. Binding connects queues to exchanges. You can think of queues like our letterbox. Exchange pushes messages into queues, from which interested consumers can consume them.

An exchange can be bound to multiple queues, and a queue can be linked to numerous exchanges. The consumer might also listen to messages from multiple queues.

Connection & Channel

RabbitMQ is designed to implement the AMQP (Advance Message Queuing Protocol). AMQP is an open messaging protocol that defines the rules for message exchange, queueing, and routing in a messaging system.

To communicate with RabbitMQ, a client or application must first establish a connection. A client can be either a producer or a consumer. The connection is established using either TCP or TLS. The main purpose of a connection is to establish a secure path between the client and RabbitMQ.

A connection can have several channels. But, why do we need channels? We can set up multiple connections between the client and the broker to exchange messages. Keeping multiple TCP connections open at the same time is undesirable because it consumes system resources and makes firewall configuration more difficult. So, according to the AMQP protocol, channels are "lightweight connections that share a single TCP connection".

We can use the amqp-client Java library to communicate with RabbitMQ in a Java application. This library is available through the Maven repository. We need to install RabbitMQ on our local machine to communicate with the Java application. You can find an installation guide here.

Producer & Consumer using Java

Now, let's look at how to create a message and how to consume it using RabbitMQ and Java. First, we write a program that will connect and publish a message to RabbitMQ.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;

public class Publisher {
    private static final String HOST = "localhost";
    private static final String QUEUE = "OWL-POST";
    private static final String MESSAGE = "Happy birthday Hermione!";

    public static void main(String[] args) {
        // CREATE A CONNECTION FACTORY
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(HOST);

        // CREATE A CONNECTION FROM FACTORY
        try (Connection connection = connectionFactory.newConnection()) {
            // GET CHANNEL FROM CONNECTION
            Channel channel = connection.createChannel();

            // ASSIGN A QUEUE TO CHANNEL
            channel.queueDeclare(QUEUE, false, false, false, null);

            // PUBLISH A MESSAGE TO CHANNEL
            channel.basicPublish("", QUEUE, null, MESSAGE.getBytes(StandardCharsets.UTF_8));
            System.out.println(" Message sent : '" + MESSAGE + "'");
        } catch (Exception e) {
            System.out.println(e.getMessage() + ", " + e.getStackTrace());
        }
    }
}

We have used amqp-client Java client library to communicate with RabbitMQ.

RabbitMQ is running on our local machine, so the host is localhost. The queue name is 'OWL-POST'. Consumer should connect to this queue to consume message from producer.

ConnectionFactory is a factory class that allows you to open a connection to RabbitMQ.

First, we get a connection from ConnectionFactory, and then we create a channel. We create a queue using queueDeclare() method by passing the queue name. This queue is idempotent, which means that if it already exists under this name, it will not be created again.

Next, we use the basicPublish() method to send our encoded message to the queue. The first argument of the basicPublish() method is the exchange name. Because we passed a blank string as the first argument, we are connecting to RabbitMQ's default exchange.

Here is our consumer.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;

public class Consumer {
    private static final String HOST = "localhost";
    private static final String QUEUE = "OWL-POST";

    public static void main(String[] args) {
        // CREATE A CONNECTION FACTORY
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(HOST);

        try {
            // CREATE A CONNECTION
            Connection connection = connectionFactory.newConnection();

            // CREATE A CHANNEL FROM CONNECTION
            Channel channel = connection.createChannel();

            // ASSIGN A QUEUE TO CHANNEL
            channel.queueDeclare(QUEUE, false, false, false, null);
            System.out.println(" Waiting for message...");

            DeliverCallback deliverCallback = (consumeMsg, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println(" Message received : '" + message + "'");
            };

            channel.basicConsume(QUEUE, true, deliverCallback, consumeMsg -> {});
        } catch (Exception e) {
            System.out.println(e.getMessage() + ", " + e.getStackTrace());
        }
    }
}

So the consumer is connected to the 'OWL-POST' queue. And it receives the message via the basicConsume() method.

DeliverCallback is a callback interface. It is notified when the consumer receives the message and is passed as an argument to the basicConsume() method.

Here is the message from Harry -'Happy birthday Hermione!'.

Source code: Publisher and Consumer.

Happy coding!!! 😊