Working with RabbitMQ

Photo by Ансплэш Степана on Unsplash

Hi everyone, it’s been a long time that the last time I was active on medium. Thank you so much for always follow my writing journeys. In this post I just want to share about RabbitMQ, an open source message broker that implemented Advanced Message Queuing Protocol. In this post I will use Go as the programming language for the example.

RabbitMQ is a reliable and mature messaging and streaming broker, which is easy to deploy on cloud environments, on-premises, and on your local machine.

Instalation

You can install RabbitMQ on your device by following these official documentation from RabbitMQ through this link. I will use the docker image for this purpose. After installation complete I recommend to enable the web management by run rabbitmq-plugins enable rabbitmq_management then you will be able to access the web management through localhost:15672/ , you can login with guest as username and password.

Terms we need to know

RabbitMQ is a message broker: it accepts and forwards messages. You can think about it as a post office: when you put the mail that you want posting in a post box, you can be sure that the letter carrier will eventually deliver the mail to your recipient. In this analogy, RabbitMQ is a post box, a post office, and a letter carrier.

Based on that statement, we need to know about:

1. Producer, is a program that send messages
2. Channel, is a virtual connection inside a connection
3. Exchange, received messages from producer and will push them to the queue
4. Queue, is a buffer that stores messages
5. Consumer, is a program mostly waits to receive messages

Simple Pub/Sub

We will create a simple pub/sub using Go (Programming language) with RabbitMQ, we will create a scenario that 1 service as a producer will broadcast messages to all related services (consumer)

• open the rabbitmq web management and login with your account (use guest as username and password if you have one)
• go to the exchanges tab and add new exchanges, for this post we will use fanout type (an exchange type that send the messages to all destination)

Add exchanges

• go to queues and streams tab and add new queue, I will use the Quorum type

Add queue

• I will create 1 queue with the same process as above and it’s name will be sms
• bind the queue with the exchanges that we have already define before

Bind queue with exchanges

• now we will create a simple apps with Go, first make a directory go-rabbitmq (or whatever it’s name would you like to)
• change to go-rabbitmq directory then run go mod init <your_module_name>
• run go get github.com/rabbitmq/amqp091-go
• then I will create folder structure will be something like this

Go structure

• on the rabbitmq folder will handle the dial up to the RabbitMQ, make rabbitmq.go then the code will be something like this

package rabbitmq

import (
 amqp "github.com/rabbitmq/amqp091-go"
)

func Connect(url string) (*amqp.Connection, error) {
 return amqp.Dial(url)
}

• create main.go on the producer folder

package main

import (
 "context"

 "github.com/awgst/go-rabbitmq/rabbitmq"
 amqp "github.com/rabbitmq/amqp091-go"
)

const rmqUrl = "amqp://guest:guest@localhost:5672/"

func main() {
 // Connect to rabbitmq
 rmqConn, err := rabbitmq.Connect(rmqUrl)
 if err != nil {
  panic(err)
 }
 defer rmqConn.Close()

 // Open a channel
 rmqChannel, err := rmqConn.Channel()
 if err != nil {
  panic(rmqChannel)
 }

 // Publish a message
 err = rmqChannel.PublishWithContext(
  context.Background(),
  "notifications", // exchange
  "",              // routing key
  false,           // mandatory
  false,           // immediate
  amqp.Publishing{
   ContentType: "text/plain",
   Body:        []byte("Hello World!"),
  },
 )
 if err != nil {
  panic(err)
 }

}

• create main.go on the consumer1 folder

package main

import (
 "fmt"

 "github.com/awgst/go-rabbitmq/rabbitmq"
)

const rmqUrl = "amqp://guest:guest@localhost:5672/"

func main() {
 // Connect to the rabbitmq
 rmqConn, err := rabbitmq.Connect(rmqUrl)
 if err != nil {
  panic(err)
 }

 // Open channel
 rmqChannel, err := rmqConn.Channel()
 if err != nil {
  panic(err)
 }

 // Consume messages from email queue
 messages, err := rmqChannel.Consume(
  "email", // queue
  "",      // consumer
  true,    // auto-ack
  false,   // exclusive
  false,   // no-local
  false,   // no-wait
  nil,     // args
 )
 if err != nil {
  panic(err)
 }

 fmt.Println("Start to listen messages...")
 for msg := range messages {
  fmt.Printf("Message body: %s", msg.Body)
 }
}

• on the consumer2 folder the main.go value is the same with the consumer1 but the difference only it will consume from sms queue while the consumer1 consume from email queue
• run all the consumers and producer

publish messages from producer

consume messages for the consumer1

consumer messages for the consumer2

I think that’s all that I could share with you for this post, thank you for read my post. Please feel free to leave any comments if you have any questions or suggestions. Share this post if you find this is useful.

See ya

References
- https://www.rabbitmq.com/
- https://www.youtube.com/watch?v=sJvb3ALAmzc&t=1770s

Source code: https://github.com/awgst/go-rabbitmq