Building Microservices in Go: Events and Background jobs using RabbitMQ
Tags: go, rabbitmq, microservices
Published: 2021-08-20 05:05:37

Originally published at [mariocarrion.com](https://mariocarrion.com/2021/05/28/golang-microservices-events-background-jobs-rabbitmq.html)

## What is RabbitMQ?

[RabbitMQ](https://www.rabbitmq.com/) is a message-queueing software also known as a message broker or queue manager. It supports protocols such as [AMQP](https://www.amqp.org/about/what), [MQTT](https://mqtt.org/) and [STOMP](https://stomp.github.io/), to name a few. RabbitMQ could be used for long-running tasks, for example background jobs, and for communication between different services.

## How does RabbitMQ work?

The easiest analogy to describe RabbitMQ is that of a Post Office and the required steps involved, from beginning to end, to deliver a mail to the final destination. In real life those steps consist of dropping off the mail into a mailbox, then some processing behind the scenes to route that mail and finally a mail person brings that mail to the destination.

![How does RabbitMQ Work? 1](https://mariocarrion.com/img/2021/05/28/1.png?src=devto)

RabbitMQ works as a queue where _Publishers_ can submit messages which then are eventually consumed by multiple _Consumers_; however the interesting part about RabbitMQ is the intermediary mechanism that sits between those publishers and consumers. This intermediary is called _Exchange_, this Exchange can be configured to define _Bindings_ to allow those messages to be routed into different _Queues_ which then the clients can listen to for consuming messages in different ways, to perhaps consume by a only concrete key or a wildcard.

![How does RabbitMQ Work? 2](https://mariocarrion.com/img/2021/05/28/2.png?src=devto)

## Publisher implementation using a Repository

> The code used for this post is [available on Github](https://github.com/MarioCarrion/todo-api-microservice-example/tree/b886f4346851c860728ec7924d8b3196139af11e).

To interact with RabbitMQ we will use the package [`streadway/amqp`](https://github.com/streadway/amqp) and similar to other data stores we will be defining a [Repository](https://mariocarrion.com/2021/04/04/golang-microservices-repository-pattern.html) that will be interacting with the actual RabbitMQ publisher and will be called in the `service` type.

This repository type will be named [rabbitmq.Task](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/internal/rabbitmq/task.go), it will contain an unexported field referencing a RabbitMQ channel and the corresponding methods required for emitting three events `Created`, `Deleted` and `Updated`:

```go
func (t *Task) Created(ctx context.Context, task internal.Task) error {
	return t.publish(ctx, "Task.Created", "tasks.event.created", task)
}

func (t *Task) Deleted(ctx context.Context, id string) error {
	return t.publish(ctx, "Task.Deleted", "tasks.event.deleted", id)
}

func (t *Task) Updated(ctx context.Context, task internal.Task) error {
	return t.publish(ctx, "Task.Updated", "tasks.event.updated", task)
}
```

Those three methods will refer to an unexported method called `publish`, which is used for publishing the data, this data is the result encoding the message using the `encoding/gob` package, similar to the code used when we discussed [Caching with Memcached](https://mariocarrion.com/2021/01/30/tips-building-microservices-in-go-golang-caching-memcached.html):

```go
func (t *Task) publish(ctx context.Context, spanName, routingKey string, e interface{}) error {
	// XXX: Excluding OpenTelemetry and error checking for simplicity
	var b bytes.Buffer

	_ = gob.NewEncoder(&b).Encode(e)

	_ = t.ch.Publish(
		"tasks",    // exchange
		routingKey, // routing key
		false,      // mandatory
		false,      // immediate
		amqp.Publishing{
			AppId:       "tasks-rest-server",
			ContentType: "application/x-encoding-gob",
			Body:        b.Bytes(),
			Timestamp:   time.Now(),
		})

	return nil
}
```

Next, in the `service` package the [`service.Task`](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/internal/service/task.go#L40-L46) type is updated to receive an instance of that repository using an interface type, which then will be used it _after_ the persistent datastore call is done, something like:

```go
func (t *Task) Create(ctx context.Context, description string, priority internal.Priority, dates internal.Dates) (internal.Task, error) {
	// XXX: Excluding OpenTelemetry and error checking for simplicity
	task, _ := t.repo.Create(ctx, description, priority, dates)

	_ = t.msgBroker.Created(ctx, task)

	return task, nil
}
```

Please refer to the [`Delete` call](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/internal/service/task.go#L78-L92) as well as the [`Update` call](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/internal/service/task.go#L108-L128) for more details, in practice the code is similar to the one above.

Now, let's take a look at the subscriber implementation. For this example we will implement a new running process in charge of consuming that data.

## Subscriber implementation

This [new process](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/cmd/elasticsearch-indexer/main.go) will consume those RabbitMQ events to properly index the Task records, changing the way we were using [Elasticsearch](https://mariocarrion.com/2021/05/24/golang-microservices-searching-with-elasticsearch.html) originally, it will also support [Graceful shutdown](https://mariocarrion.com/2021/05/21/golang-microservices-graceful-shutdown.html) like we previously covered.

The code for listening, at the moment, is a slightly long [method](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/cmd/elasticsearch-indexer/main.go#L137-L226), the [important part](https://github.com/MarioCarrion/todo-api-microservice-example/blob/b886f4346851c860728ec7924d8b3196139af11e/cmd/elasticsearch-indexer/main.go#L176-L223) would be the actual Go channel returned by RabbitMQ, this code does something like the following to receive all events:

```go
// XXX: Excluding some things for simplicity, please refer to the original code
for msg := range msgs {
	switch msg.RoutingKey {
	case "tasks.event.updated", "tasks.event.created":
		// decode received Task event
		// call Elasticsearch to index record
	case "tasks.event.deleted":
		// decode received Task event
		// call Elasticsearch to delete record
	}
	// acknowledege received event
}
```

In a real-life implementation you should consider implementing a _Server_ type able to handle different events, perhaps similar to the way `net/http.Server` works and maybe define something similar to a Muxer to allow listening to multiple events with their corresponding encoding/decoding logic.

## Conclusion

_RabbitMQ_ is commonly known as a distributed queue but it can also be used as a message broker to communicate multiple services, is a powerful tool that thanks to the available configuration options could serve to deliver messages to multiple clients at scale.

## Recommended Reading

If you're looking to do something similar in Kafka and Redis, I recommend reading the following links:

* [Microservices in Go: Events Streaming using Kafka](https://mariocarrion.com/2021/06/03/golang-microservices-events-streaming-kafka.html)
* [Microservices in Go: Using Pub/Sub with Redis](https://mariocarrion.com/2021/06/10/golang-microservices-pub-sub-redis.html)

---

> This post includes icons made by [itim2101](https://www.flaticon.com/authors/itim2101) from [Flaticon](https://www.flaticon.com/)