Microservices & RabbitMQ On Docker
Tags: python, go, microservices, rabbitmq
Published: 2018-04-25 23:59:44

---
title: Microservices & RabbitMQ On Docker
published: true
description: 
tags: python, go, microservices, rabbitmq
cover_image: https://thepracticaldev.s3.amazonaws.com/i/peq82ldxda01dxo7up77.png
---

A microservices-based architecture involves decomposing your monolith app into multiple, totally *independently deployable and scalable services*. Beyond this base definition, what constitutes a microservice can be somewhat subjective, though there are several battle-tested practices adopted by giants like [Netflix](https://www.nginx.com/blog/microservices-at-netflix-architectural-best-practices/) and [Uber](https://eng.uber.com/building-tincup/) that should always be considered. And I'll discuss some of them. Ultimately, we want to *divide our app into smaller apps, each of which is a system apart & deals with only one aspect of the whole app and does it really well*. This decomposition is a very consequential step and can be done on the basis of [subdomains](http://microservices.io/patterns/decomposition/decompose-by-subdomain.html), which have to be identified correctly. The smaller apps are more **modular & manageable** with well-defined boundaries, can be written using **different languages/frameworks**, **fail in isolation** so that the entire app doesn't go down (no SPOF). Take a Cinema ticketing example:

![](https://thepracticaldev.s3.amazonaws.com/i/m4cppxfwj2g8ml4uqqxr.png)
<sub><sup>*Source: https://codeburst.io/build-a-nodejs-cinema-api-gateway-and-deploying-it-to-docker-part-4-703c2b0dd269*</sup></sub>

Let's break down this bird's eye view:

i) The user app can be a mobile client, SPA etc or any client consuming our backend services.

ii) **It's considered a bad practice to ask our clients to communicate with each of our services separately**, for reasons I tried to explain [here](https://dev.to/djviolin/how-do-you-protecting-your-backend-api-in-your-microservice-if-you-use-a-single-page-application-on-the-frontend-2ie1/comments/2jj9). This is what the API gateways are for: to receive client requests, call our service(s), return a response. Thus the client only has to talk to one server, giving the illusion of a monolith. Multiple gateways may be used for *different kinds of clients* (mobile apps, tablets, browsers etc.). They can and should be responsible for limited functionality like merging/joining responses from services, authentication, ACLs. In large applications, which need to scale and move dynamically, gateways also need access to a [Service Registry](http://microservices.io/patterns/service-registry.html) which holds the locations of our microservice instances, databases etc.

iii) **Each service has its own storage**. This is a key point and ensures loose coupling. Some queries will then need to join data that is owned by multiple services. To avoid this major performance hit, data may be replicated and sharded. This principle is [not just tolerated in microservices, but encouraged](http://microservices.io/patterns/data/database-per-service.html).

iv) The **REST calls made to our API gateway** are passed to the services, which in turn talk to other services, return a result to the gateway which, perhaps, compiles it and responds with it to the client. Communication among services like this on one client request to the app should not happen. Otherwise we'll be sacrificing performance, on account of another HTTP round-trip, for the newly introduced modularity.

**A single request should, ideally, only invoke one service to fetch the response**. This means any synchronous requests between the services should be minimized, and that's not always possible; mechanisms like [gRPC](https://grpc.io/), [Thrift](https://thrift.apache.org/) or even simple HTTP (as in our example) are commonly employed when necessary. As you may have guessed, this implies that data will have to be replicated across our services. Say, the `GET /catalog/<<cityId>>` endpoint is also supposed to return the premieres at each cinema in the city at that time. With our new strategy, the premieres will have to be stored in the database for the *Cinema Catalog* service as well. Hence, point `iii)`.


##Communicating Asynchronously Between The Services

So, say, the premieres change as a result of some CRUD operation on the *Movies* service. To keep the data in sync, that update event will have to be emitted and applied to the *Cinema Catalog* service as well. **Try to picture our microservices as a cluster of state machines where updates in states may have to be communicated across the cluster to achieve eventual consistency**. Of course, we should never expect our end-users to have to wait longer for requests to finish and sacrifice their time for modularity to our benefit. Thereby, all of this communication has to be non-blocking. And that's where [RabbitMQ](https://www.rabbitmq.com) comes in.

RabbitMQ is a very [powerful](https://www.rabbitmq.com/#features) message broker that implements the [AMQP messaging protocol](https://www.rabbitmq.com/amqp-0-9-1-quickref.html). Here's the abstract: first, you install a RabbitMQ server instance (broker) on a system. Then a *publisher/producer* program connects to this server and sends out a message. RabbitMQ queues that message and siphons it off to a single or multiple *subscriber/consumer* programs that are out there listening on the RabbitMQ server.

Before I get to the crux of this article, I want to explicitly declare that microservices are *way* more complex and we won't be covering critical topics like [fault tolerance](https://blog.risingstack.com/designing-microservices-architecture-for-failure/) because of the intricacy of distributed systems, the full [role of the API gateway](http://microservices.io/patterns/apigateway.html), [Service Discovery](http://microservices.io/patterns/client-side-discovery.html), data consistency patterns like [Sagas](http://microservices.io/patterns/data/saga.html), preventing service failure cascading using [Circuit Breakers](http://microservices.io/patterns/reliability/circuit-breaker.html), health checks and architectural patterns like [CQRS](https://martinfowler.com/bliki/CQRS.html). Not to mention [how to decide whether microservices will work for you or not](https://martinfowler.com/articles/microservice-trade-offs.html).


##How RabbitMQ Works

More specifically, the messages are *published* to an *exchange* inside the RabbitMQ broker. The exchange then distributes copies of that message to *queues* on the basis of certain developer-defined rules called *bindings*. This part of the messages' journey is called *routing*. And this indirection is of course what makes for the non-blocking message transmissions.  *Consumers* listening on those *queues* that got the message will receive it. Pretty simple, right?

Not exactly. There are [four different types of *exchanges*](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges) and each, along with the *bindings*, defines a routing algorithm. "Routing algorithm" means, essentially, how the messages are distributed among the queues. Going into the details about each type might be an overkill here, so I'll just expand on the one we'll be using: the [*topic exchange*](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchange-topic):

For an exchange to push a message onto a queue, that queue must be bound to the exchange. We can create multiple exchanges with unique names, explicitly. However, when you deploy RabbitMQ it comes with a default, nameless exchange. And every queue we create will be automatically bound to this exchange. To be descriptive, I'll be creating a named exchange manually and then bind a queue to it. This binding is defined by a *binding key*. The exact way a binding key works, again, depends on the type of the exchange. Here's how it works with a topic exchange:

- A *queue* is bound to an *exchange* using a *string pattern* (the binding key)
- The published message is delivered to the *exchange* along with a *routing key*
- The *exchange* checks which *queues* match the *routing key* based on the *binding key* pattern defined before.

<sub><sup>`*` can substitute for exactly one word. `#` can substitute for zero or more words.<sup/><sub/>
![](https://thepracticaldev.s3.amazonaws.com/i/zepcpbrn07266rcc71sc.png)
<sub><sup>*Source: https://www.rabbitmq.com/tutorials/tutorial-five-python.html*<sup/><sub/>

Any message with a routing key `"quick.orange.rabbit"` will be delivered to both queues. However, messages with `"lazy.brown.fox"` will only reach `Q2`. Those with a routing key not matching any pattern will be lost.

For some perspective, let's just skim over two other exchange types:

- **Fanout exchange**: Messages sent to this kind of exchange will be sent to ALL the queues bound to it. The routing key, if provided, will be completely ignored. This can be used, for example, for broadcasting global configuration updates across a distributed system.
- **Direct exchange** (simplest): Sends the message to the queue whose binding key is *exactly* equal to the given routing key. If there are multiple consumers listening on the queue, then the messages will be load-balanced among them, hence, it is commonly used to distribute tasks between multiple workers in a round robin manner.


My illustration will be *very* simple: a Python [Flask](http://flask.pocoo.org/) app with a single POST endpoint, which, when called, will purport to update a user's info, emit a message to the RabbitMQ broker (non-blocking of course) and return a 201. A separate Go service will be listening for the message from the broker and hence have the chance to update its data accordingly. All three will be hosted on separate containers. 



##Setting Up Our Containerized Microservices & Broker Using Docker Compose 

Provisioning a bunch of containers and all that goes along with them can be a pain, so I always rely on [Docker Compose](https://docs.docker.com/compose/).

Here's [the entire code](https://github.com/UsamaAshraf/microservices-using-rabbitmq). We're declaring three services that will be used for the three containers. The two [volumes](https://docs.docker.com/storage/volumes/) are needed for putting our code inside the containers:

```yml
# docker-compose.yml

version: "3.2"
services:
    rabbitmq-server:
        build: ./rabbitmq-server

    python-service:
        build: ./python-service
        # 'rabbitmq-server' will be available as a network reference inside this service 
        # and this service will start only after the RabbitMQ service does.
        depends_on:
            - rabbitmq-server
        # Keep it running.  
        tty: true
        # Map port 3000 on the host machine to port 3000 of the container.
        # This will be used to receive HTTP requests made to the service.
        ports:
            - "3000:3000"
        volumes:
            - './python-service:/python-service'

    go-service:
        build: ./go-service
        depends_on:
            - rabbitmq-server
        tty: true
        volumes:
            - './go-service:/go-service'

# Host volumes used to store code.
volumes:
    python-service:
    go-service:
```


The Dockerfiles are pretty much the standard ones from [Docker Hub](https://hub.docker.com/), to which I've added:

- `/go-service` working directory in the Go service container.
- `/python-service` working directory in the Python service container.
- Go's RabbitMQ client library called [amqp](https://github.com/streadway/amqp)
- Python's RabbitMQ client [Pika](https://pypi.org/project/pika/0.11.0/) & [Flask](http://flask.pocoo.org/)

Our Flask app has just one endpoint that receives a `user_id` and a `full_name`, which will be used to update the user's profile. A message indicating this update will then be sent to the RabbitMQ broker.

```py
# main.py

from flask import Flask
from flask import request
from flask import jsonify
from services.user_event_handler import emit_user_profile_update

app = Flask(__name__)

@app.route('/users/<int:user_id>', methods=['POST'])
def update(user_id):
    new_name = request.form['full_name']

    # Update the user in the datastore using a local transaction...
    
    emit_user_profile_update(user_id, {'full_name': new_name})

    return jsonify({'full_name': new_name}), 201
```

The logic for emitting events to other services should always be separate from the rest of the app, so I've extracted it to a module. The *exchange* should be explicitly checked for and created by the publisher and consumer both, because we can't know (nor should we rely on) which service starts first. This is a good practice, which most RabbitMQ client libraries facilitate seamlessly: 

```py
# services/user_event_handler.py

import pika
import json

def emit_user_profile_update(user_id, new_data):
    # 'rabbitmq-server' is the network reference we have to the broker, 
    # thanks to Docker Compose.
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='rabbitmq-server'))
    channel    = connection.channel()

    exchange_name = 'user_updates'
    routing_key   = 'user.profile.update'

    # This will create the exchange if it doesn't already exist.
    channel.exchange_declare(exchange=exchange_name, exchange_type='topic', durable=True)
    
    new_data['id'] = user_id

    channel.basic_publish(exchange=exchange_name,
                          routing_key=routing_key,
                          body=json.dumps(new_data),
                          # Delivery mode 2 makes the broker save the message to disk.
                          # This will ensure that the message be restored on reboot even  
                          # if RabbitMQ crashes before having forwarded the message.
                          properties=pika.BasicProperties(
                            delivery_mode = 2,
                        ))

    print("%r sent to exchange %r with data: %r" % (routing_key, exchange_name, new_data))
    connection.close()
```
Don't get confused by `channel`. A *channel* is just a [virtual, lightweight connection](https://www.rabbitmq.com/tutorials/amqp-concepts.html#amqp-channels) *within* the TCP connection that is meant to prevent opening multiple, expensive TCP connections. Especially in multithreaded environments.

The `durable` parameter ensures that the *exchange* is persisted to the disk and can be restored if the broker crashes or goes offline for any reason. The publisher (Python service) creates an *exchange* named `user_updates` and sends it the user's updated data with `user.profile.update` as the *routing key*. This will be matched with a `user.profile.*` *binding key*, which our Go service will define:

```go
// main.go

package main

import (
	"fmt"
	"log"
	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		// Exit the program.
		panic(fmt.Sprintf("%s: %s", msg, err))
	}
}

func main() {
	// 'rabbitmq-server' is the network reference we have to the broker, 
	// thanks to Docker Compose.
	conn, err := amqp.Dial("amqp://guest:guest@rabbitmq-server:5672/")
	failOnError(err, "Error connecting to the broker")
	// Make sure we close the connection whenever the program is about to exit.
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	// Make sure we close the channel whenever the program is about to exit.
	defer ch.Close()
	
	exchangeName := "user_updates"
	bindingKey   := "user.profile.*"

	// Create the exchange if it doesn't already exist.
	err = ch.ExchangeDeclare(
			exchangeName, 	// name
			"topic",  		// type
			true,         	// durable
			false,
			false,
			false,
			nil,
	)
	failOnError(err, "Error creating the exchange")
	
	// Create the queue if it doesn't already exist.
	// This does not need to be done in the publisher because the
	// queue is only relevant to the consumer, which subscribes to it.
    // Like the exchange, let's make it durable (saved to disk) too.
	q, err := ch.QueueDeclare(
			"",    // name - empty means a random, unique name will be assigned
			true,  // durable
			false, // delete when the last consumer unsubscribes
			false, 
			false, 
			nil,   
	)
	failOnError(err, "Error creating the queue")

	// Bind the queue to the exchange based on a string pattern (binding key).
	err = ch.QueueBind(
			q.Name,       // queue name
			bindingKey,   // binding key
			exchangeName, // exchange
			false,
			nil,
	)
	failOnError(err, "Error binding the queue")

	// Subscribe to the queue.
	msgs, err := ch.Consume(
			q.Name, // queue
			"",     // consumer id - empty means a random, unique id will be assigned
			false,  // auto acknowledgement of message delivery
			false,  
			false,  
			false,  
			nil,
	)
	failOnError(err, "Failed to register as a consumer")


	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf("Received message: %s", d.Body)

			// Update the user's data on the service's 
			// associated datastore using a local transaction...

			// The 'false' indicates the success of a single delivery, 'true' would
			// mean that this delivery and all prior unacknowledged deliveries on this
			// channel will be acknowledged, which I find no reason for in this example.
			d.Ack(false)
		}
	}()
	
	fmt.Println("Service listening for events...")
	
	// Block until 'forever' receives a value, which will never happen.
	<-forever
}
```

RabbitMQ uses port 5672 by default for non-TLS connections and "guest" as the username & password. You can study the [plethora of configuration options](https://www.rabbitmq.com/configure.html) available and how to use them with [Pika](https://pika.readthedocs.io/en/0.11.2/modules/index.html) and [Go amqp](https://godoc.org/github.com/streadway/amqp).


You might be wondering what this line is for: **`d.Ack(false)`**

This tells the broker that the message has been delivered, processed successfully and can be deleted. By default, these acknowledgments happen automatically. But we specified so otherwise when we subscribed to the queue: **`ch.Consume`**. 

Now, if the Go service crashes (for any unforeseeable reason), the acknowledgment won't be sent and this will cause the broker to re-queue the message so that it may be given another chance to be processed.


## Launching The Microservices

Alright, let's fire'em up:

Run **`docker-compose up`**

When the three services have been built (will take at least a few minutes the first time), check for their names using **`docker ps`**:

![](https://thepracticaldev.s3.amazonaws.com/i/694axi1gm9nxudiu418g.png)

Open two new terminals, SSH into the Python and Go containers using the respective container names and start the servers:

**`docker exec -it microservicesusingrabbitmq_python-service_1 bash`**
**`FLASK_APP=main.py python -m flask run --port 3000 --host 0.0.0.0`**


**`docker exec -it microservicesusingrabbitmq_go-service_1 bash`**
**`go run main.go`**


Open a third terminal to send the POST request. I'll use Curl:

**`curl -d "full_name=usama" -X POST http://localhost:3000/users/1`**

And you'll see the transmission:

![](https://thepracticaldev.s3.amazonaws.com/i/9945lbxplrxsm5e21qsp.png)

![](https://thepracticaldev.s3.amazonaws.com/i/ufwdryd0l0zw53k11wm2.png)


At any point, you may also SSH into the RabbitMQ container and just look around:

- `rabbitmqctl list_exchanges` (list all the exchanges on this broker node)
- `rabbitmqctl list_queues` (list all the queues on this broker node)
- `rabbitmqctl list_bindings` (list all the bindings on this broker node)
- `rabbitmqctl list_queues name messages_ready messages_unacknowledged` (list all the queues with the number of messages each has that are *ready to be delivered to clients but not yet delivered* and those *delivered but not yet acknowledged*)


As I mentioned at the start, this is by no means a deep dive into microservices. There are many questions to be asked and I'll try to answer an important one: how do we make this communication transactional? **So what happens if our Go service (consumer) throws an exception while updating the state on its end and we need to make sure that the update event rolls back across all the services that were affected by it?**. Imagine how complicated this could get when we have several microservices and thousands of such "update events". **Essentially, we'll need to incorporate separate events that perform the rollback**.

In our case, if the Go service throws an exception while updating the data, it will have to send a message back to the Python service telling it to rollback the update. *It's also important to note that in the case of these kinds of errors, the message delivery will have to be acknowledged (even though the processing wasn't successful), so that the message doesn't get re-queued by the broker*. When writing our consumer, we'll have to decide which errors mean that the message should be re-queued (tried again) and which mean that the message should not be re-queued and just rolled back.

But how do we specify which update event to rollback and how exactly would the rollback happen? The [Saga pattern](https://blog.couchbase.com/saga-pattern-implement-business-transactions-using-microservices-part/) along with [event sourcing](http://microservices.io/patterns/data/event-sourcing.html) is widely used to ensure such data consistency.


## A Few Words On Designing The Broker

Consider two things: the **types of exchanges to use** and **how to group the exchanges**. 

If you need to broadcast certain kinds of messages to all the services in your system, have a look at the [fanout exchange type](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchange-fanout). Then, one way to group the exchanges could be on the basis of events e.g. three fanout exchanges named `user.profile.updated`, `user.profile.deleted`, `user.profile.added`. This might not be what you want all the time since you might end up with too many exchanges and won’t be able to filter messages for specific consumers without creating a new *exchange*, *queue* and *binding*.

Another way could be to create *topic exchanges* in terms of entities in your system. So in our first example, `user`, `movie`, `cinema` etc could be entities and, say, *queues* bound to `user` could use binding keys like `user.created` (get message when a user is created), `user.login` (get message when a user has just logged in), `user.roles.grant` (get message telling that the user has been given an authorization role), `user.notify` (send the user a notification) etc.

Always use routing to filter and deliver messages to specific consumers. Writing code to discard certain incoming messages while accepting others is an anti-pattern. Consumers should only receive messages that they require.

Finally, if your needs are complex and you require that messages be filtered down to certain consumers on the basis of multiple properties, use a [headers exchange](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchange-headers).


[*Enjoy!*](https://github.com/UsamaAshraf/microservices-using-rabbitmq)