Fanout Async Messaging Design In Microservices Using RabbitMQ

Introduction

 

Asynchronous messaging pattern is an indispensable design choice when it comes to microservice communications. In this design pattern, one microservice can publish a message and doesn't need to wait for completion of processing  by the consumer to proceed with its work.

 

There are various techniques that could be implemented to achieve asynchronous interactions among microservices. In this article, I shall talk about a use case where we would be using a message broker – RabbitMQ – and “Fanout” technique to implement the asynchronous messaging pattern in action.

 

What is the “Fanout” Messaging Pattern?

 

Fanout is a messaging design where the published message from a particular publisher is consumed by multiple different subscribers independently and simultaneously. The intention is that the same published message will be consumed by different consumers and be processed in different ways.

 

In a message broker like RabbitMQ, the publisher publishes a message which is put into different queues through a special type of exchange called “Fanout Exchange”.

 

Consumers listening to these queues get the same message to be consumed.

 

The below diagram shows a diagrammatic representation of “Fanout” messaging pattern.

 

 

When to Use Fanout Pattern?

 

The use case of fanout pattern is applicable in places where a publisher needs to asynchronously communicate to multiple consumers on a single workload.

An example could be: Let’s say in an online library system, a microservice is responsible for placing the borrowing request for a book – Book Rent service. Here, once a book is successfully rented out to a customer, this service can act as a publisher and produce a message for other services that are responsible for (i) Removing this from the customer’s ‘Wishlist’, (ii) Reducing the availability count for this book in the online library shelf, (iii) Updating the “Currently Reading Shelf” of the customer. These consumer services would listen for the same message in different queues and can work on this notification independently without making the renting process to wait. Here the publisher, “fans out” the message in different queues where interested consumers are hooked into it for consuming the message. The publisher here, the Book Rent service, doesn't need to wait until other services are finished acting on the message and can seamlessly and independently issue the book to the customer and carry out more renting requests making it available for the users.

 

Refresher on “Exchanges”, “Bindings”, “Routing Keys” on RabbitMQ

 

As a refresher or quick reckoner, I am delineating the different key aspects of the RabbitMQ message broker that we would be using in the prototype. However, I encourage you to go for further readings to have in-depth understanding on these topics.

 

Exchanges

 

Exchange is the place where all the published messages first land from publishers and then get routed to different queues. This could be compared to a mailbox or a hub where all the messages arrive and then based on the routing key, header attribute and binding are forwarded to respective queues.

 

Following are the types of exchanges supported in RabbitMQ,

• Direct
• Fanout
• Topic
• Headers

Bindings

 

Binding links a queue to an exchange.

 

Routing Keys

 

Routing Key is one of the attributes that the exchanges look for while deciding which queue the message will be routed to. However, in the case of a “Fanout” type of exchange this key doesn’t have any significance and remains ignored.

 

Prototype

 

In my earlier article I had demonstrated how to spin-up and communicate to a RabbitMQ container. This demo would be based on the same infrastructure.

 

To begin, (A) I would create an exchange of type “Fanout” in my running RabbitMQ instance. After that (B) I would declare three different queues to be bound to this exchange. Once this is ready, (C) I shall publish a message to the exchange through the rent-out service.

 

Below code snippets shows these steps, 

private const string FANOUT_EXCHANGE = "fanout.exchange";
private const string WISHLIST_QUEUE = "wishListQueue";
private const string LIBRARY_SHELF_QUEUE = "libraryShelfQueue";
private const string READING_SHELF = "readingShelfQueue";

var factory = new ConnectionFactory()
            {
                Uri = new Uri("amqp://guest:guest@localhost:5672")
            };

            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                // declare exchange of type "Fanout"
                channel.ExchangeDeclare(exchange:FANOUT_EXCHANGE, type: ExchangeType.Fanout);

                // declare queue : Wishlist Queue
                channel.QueueDeclare(queue: WISHLIST_QUEUE,
                                     durable: false,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
                // bindind the wishlist queue to the "Fanout" exchange
                channel.QueueBind(queue: WISHLIST_QUEUE, exchange: FANOUT_EXCHANGE, routingKey:"");

                // declare queue : Library Shelf Queue
                channel.QueueDeclare(queue: LIBRARY_SHELF_QUEUE,
                                     durable: false,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
                // bindind the Library Shelf Queue to the "Fanout" exchange
                channel.QueueBind(queue: LIBRARY_SHELF_QUEUE, exchange: FANOUT_EXCHANGE, routingKey: "");

                // declare queue : Reading Shelf Queue
                channel.QueueDeclare(queue: READING_SHELF,
                                     durable: false,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
                // bindind the Reading Shelf to the "Fanout" exchange
                channel.QueueBind(queue: READING_SHELF, exchange: FANOUT_EXCHANGE, routingKey: "");




                // publish message
                string message = "ID of the Book which is Rented out: " + id;
                var body = Encoding.UTF8.GetBytes(message);

                channel.BasicPublish(exchange: FANOUT_EXCHANGE,
                                    routingKey: "",
                                    basicProperties: null,
                                    body: body);

            }

With this code executed, when I check in the RabbitMQ management console, I should see the following:

A new exchange with the given name is created and being listed,

 

Three new queues are created which are now bound to that new Fanout exchange,

 

 

The single new message has been queued in three different queues as per the “Fanout” pattern,

 

Now let us spin up our consumers who are configured to listen to these three different queues. A sample code from one of the consumers – WishListService – where it consumes the messages from the respective queue: wishListQueue

//establish connection
            var factory = new ConnectionFactory()
            {
                Uri = new Uri("amqp://guest:guest@localhost:5672")
            };
            var rabbitMqConnection = factory.CreateConnection();
            var rabbitMqChannel = rabbitMqConnection.CreateModel();

            rabbitMqChannel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

            //consume the message received
            var consumer = new EventingBasicConsumer(rabbitMqChannel);
            consumer.Received += (model, args) =>
            {
                var body = args.Body;
                var message = Encoding.UTF8.GetString(body.ToArray());
                Console.WriteLine("Book ID To be removed from Wishlist: " + message);
                rabbitMqChannel.BasicAck(deliveryTag: args.DeliveryTag, multiple: true);
                Thread.Sleep(1000);
            };
            rabbitMqChannel.BasicConsume(queue: WISHLIST_QUEUE,
                                         autoAck: false,
                                         consumer: consumer);
            Console.ReadLine();

Once we run these consumers, the output looks like this,

 

 

In this prototype, I am sending a Book ID to the publisher service through a Web API call which publishes a message with the book ID in the message broker fanout exchange and then the following consumer services are consuming the message to process the workload independently,

1. WishListService
2. LibraryShelfService
3. CurrentReadingShelfService

Here is the complete snapshot that shows all these working in a single image,

 

 

Summary

 

In this article, we first understood the Fanout async messaging pattern which could be used in Microservice communications. We then started exploring how this could be achieved through a message broker such as RabbitMQ. For that we created a prototype based on our hypothetical online digital library system where the renting service publishes a single message upon successfully renting out a book. This message is published through a Fanout exchange in RabbitMQ which places it in different bound queues. Consumer services listening to the respective queue pick up this message and process them independently. This way we ensure our Renting service is not blocked for any independent work to be completed and free for picking up the next request for renting out while other services perform their work independently on each rent request.