Building an Event Bus with RabbitMQ
An Event bus is a mechanism that allows different components to communicate with each other without knowing about each other. A component can send an event to the Event bus without knowing who will pick it up or how many others will pick it up. Components can also listen to Events on an Event bus, without knowing who sent the Events. That way, components can communicate without depending on each other. Also, it is very easy to substitute a component. As long as the new component understands the Events that are being sent and received, the other components will never know.
In a distributed system there are mainly two modes of communication which are: asynchronous and synchronous communication. In synchronous communication when you send a request, you would need to wait for a response, while asynchronous is the opposite, you don’t get a response instantly. One way to build an event bus for a distributed system is to leverage a Message Broker for asynchronous communication. A message broker is software that provides a reliable means of communication between different applications. Examples of popular message brokers are RabbitMQ, Kafka, NAT Gateway, SQS, SNS, Google PubSub, etc. Today we are building a simple Event Bus using RabbitMQ to decouple different services and illustrate how they can send and receive messages.
RabbitMQ is lightweight and easy to deploy on-premises and in the cloud. It supports multiple messaging protocols. RabbitMQ can be deployed in distributed and federated configurations to meet high-scale, high-availability requirements. The major difference between RabbitMQ and other message brokers is its open source and lightweight. To spin up a simple RabbitMQ container is very easy and it would require less than 300MB of memory to run, or you can make use of https://www.cloudamqp.com/ to understand the basics of RabbitMQ.
How RabbitMQ works
Publisher: It is an application that sends a message to the RabbitMQ message broker. The application can be a Node, PHP, Ruby on rails, or Go app.
Exchange: The publisher sends the message to an exchange, which is responsible for routing the message to a queue. Message routing is where RabbitMQ stands out because it provides you with a wide range of options to route messages to different queues. RabbitMQ provides different types of exchanges: Direct, Fanout, Topic, Header, and Nameless exchange (Default).
Queue: Queue is the data structure that is similar to the queue in the real world. A queue is a data structure in which whatever comes first will go out first. In RabbitMQ, messages are published from an application to an exchange and are routed using a routing key to a queue. In a distributed system every service can consume messages from one or more queues.
Consumer: It’s an application that connects to a queue, and pulls all the messages sent to the queue in real time.
Designing an event bus for a simple blog.
In the diagram above, the API Gateway routes the synchronous request to the user service which is responsible for creating a new user and publishing the user:created event to the RabbitMQ responsible for routing the messages to the appropriate queues. The Post service listens to the post_queue, the Comment service listens to the comment_queue and the Notification service listens to the notification_queue. This is achieved using the pub-sub messaging pattern since RabbitMQ doesn’t come with the PubSub pattern out of the box like SNS, Google PubSub, and Kafka we can achieve this by using the fan-out exchange type.
Modeling an Event Bus for a simple blog.
The table above shows how different services publish events to the RabbitMQ broker. Using the fanout exchange to bind all the queues, every event published is delegated to the appropriate service. We can also use the exchange as the event, depending on the naming convention you choose to use. Preferably, events can be sent via the message header, and you can define different types of event handlers based on the event name.
In RabbitMQ in order to listen to messages from different exchanges, you need to assert the queue and the exchange exists, then you bind the queue to all the exchanges you want to listen to. The exchange in this context can be regarded as a topic in Pubsub-oriented messaging.
Handling race conditions in an event bus.
Processing messages in the right order really matters when handling data. In our event bus, we would be using FIFO (First in first out) pattern to process all incoming messages. message A must be processed before message B is processed to achieve consistency in our system. For instance, we have a horizontally scaled-out backend system using Kubernetes, each service has multiple pods running to achieve high availability, and all the pods running for a specific service are connected to the RabbitMQ message broker. By default RabbitMQ won’t send the message to all the running pods, it would only send the message to a single pod without making provision to handle race conditions there’s a 1% chance that post:updated event might be processed before post:created event or post:deleted might be processed before post:updated. In a blog where only one user can perform an update to the same record the chances of this occurring are very slim, but when you are building a fintech product that multiple users can deposit or withdraw from a wallet the chances are really high you run into this issue. In a monolith architecture, the best approach would have been to implement a row lock, but since the database is distributed, the issue is a bit more complicated. Record versioning is a good approach that solves this issue. Every time a record is created and updated the version column is incremented by one.
When we published our post:created and post:updated we would append the version to the event payload. We need to perform a check in the post:updated listener to verify if the new event is an increment of the event we processed last for that record. Get the version column for that row, perform an increment by 1 and if it matches the incoming event version, process the event and acknowledge the message else ignore the message. So this way that message would be resent to the consumer when it is ignored. Even if we have 10 Kubernetes pods or EC2 instances running for a service, messages would always be processed in the appropriate order. When the event version received is 1, there’s no need to perform this check.
Logging events.
Logging in an event bus is a necessity because in most message brokers after an event has been processed, it is removed from the message broker. The only way you can detect which service published a certain event is through logging. We can log events using any data store of our choice. Logging is very critical in tracing and debugging the flows of events within our system, there are also some edge cases we need to consider while logging in order to avoid creating a major vulnerability on our system. We would need to create a log filter to remove certain sensitive fields before they are stored in our data store because these fields are meant to be confidential between these services. For instance, if you are publishing an event that has a pin, password, or OTP field we can write a logic to filter all these fields before they are saved on our data store. If you would like to use the logs for resending failed messages, you can encrypt the original body and decrypt the body when you need to resend the message. Here are some of the fields that need to be logged.
Dead letter queue.
Dead letter queues are used in an event-driven system for different reasons, the most common ones are:
- They serve as a queue that holds rejected messages or messages that couldn’t be processed within a specific duration. The duration could be based on the number of times the message would be resent to the queue after message consumption fails or the message TTL expires.
- When a queue is full, the incoming messages are redirected to the dead letter queue.
Leveraging on dead letter queues helps to troubleshoot new edge cases in your system, if a message fails to process due to certain runtime errors, you can easily detect it when you examine the body of the message and fix that issue before attempting to resend it.
Check out a sample project on Github
References
