These days, applications need to have instant data access and processing capabilities. Whether it's updating real-time stock trades for financial institutions or navigating through live traffic data, the ability to process and react to data in real time is crucial.

In this tutorial, you’ll delve into the mechanics of Kafka and then integrate it with Go to develop a real-time notification system.

In order to understand this article fully, you should have prior knowledge of Goroutines, the Gin framework, and containerization tools like Docker.

Table of Contents

1. What is Kafka?
2. How to Set Up the Project Workspace
3. How to Create the User and Notification Models
4. How to Set Up the Kafka Producer
5. How to Set Up the Kafka Consumer
6. Let's Test the Real-Time Notification System
7. Wrapping Up

What is Kafka? 🤔

Kafka is a distributed event streaming platform. Initially developed by LinkedIn, Kafka was subsequently contributed to the Apache Software Foundation and made open-source. This transition marked its role as a key participant in real-time data streaming.

More than a simple communication tool, Kafka is an “event broker” — a system that controls and handles events or messages between various applications or services. It can handle massive daily event volumes as a distributed event streaming platform, guaranteeing that data is seamlessly transported and analyzed in real time.

Apart from its fundamental role as an event broker, Kafka offers durability, scalability, and fault-tolerance features. It also helps ensure that large-scale data streams are managed efficiently and reliably with very low latency.

Kafka’s core components ⚙️

Now that you’ve been acquainted with Kafka, let’s dive into the main elements that make up its architecture:

Events

An event records the fact that “something happened”. It can be thought of as a message or a piece of data representing a change or an action. In the context of our real-time notification system, you could consider an event as follows:

• Event key: “1” (representing the user ID for Emma)
• Event value: “Bruno started following you.”

Brokers

A Kafka broker is a server that runs the Kafka software and stores data. While large-scale production setups often involve multiple brokers across several machines, you’ll use a single broker setup for this tutorial.

Topics

Topics in Kafka are similar to folders in a filesystem. They represent categories under which data or events are stored. For instance, an example topic name could be "notifications".

Producers

Producers are entities that publish (write) or send messages to Kafka, such as a Go program or a service. When a producer has an event to send, it chooses a topic to address the event to.

Consumers

Consumers read and process events or messages from Kafka. After producers send messages to topics, consumers can subscribe to one or more topics to receive the messages.

Partitions

Each topic in Kafka can be further divided into partitions. Think of partitions as segments within a topic that enable Kafka to manage data more efficiently, especially in setups with multiple brokers.

You’ll stick with a basic configuration without delving into multiple partitions, but you should understand their role in larger Kafka deployments.

Consumer groups

While individual consumers handle messages from specific partitions, consumer groups manage coordination across multiple consumers.

A consumer group consists of multiple consumers collaboratively processing messages from different partitions of a topic. This ensures that each message from a partition is processed by just one consumer in the group, allowing for efficient and scalable consumption.

Think of it as a team of consumers working together, with each member responsible for messages from specific partitions, ensuring no message is overlooked.

Replicas

Replication ensures data safety. In larger Kafka deployments, storing multiple data replicas is common to help recover from unexpected failures.

You won’t be using replicas in this tutorial, but it’s beneficial to understand their significance in ensuring data durability in Kafka.

KRaft

KRaft is Kafka’s own consensus protocol introduced to eliminate the need for ZooKeeper. In short, KRaft manages metadata directly within Kafka, providing scalability, simplicity, and improved failover, among other benefits.

To tie all these components together, here’s a visual representation of Kafka’s core architecture, illustrating a broker, topics, partitions, and consumer groups:

Visualizing Kafka’s event streaming Architecture

How to Set Up the Project Workspace 👨‍💻👩‍💻

Enough theory for now! Let’s get our hands dirty with the actual project.

Assuming that you have Docker and Go installed on your machine, let’s create a directory for the project named kafka-notify. Then, you will pull Bitnami’s Kafka Docker image for the Kafka setup, providing a hassle-free installation:

mkdir kafka-notify && cd kafka-notify

curl -sSL \
https://raw.githubusercontent.com/bitnami/containers/main/bitnami/kafka/docker-compose.yml > docker-compose.yml

Before starting the Kafka broker, there’s a slight modification required in the docker-compose.yml file. Find the following string:

KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://:9092

And replace it with:

KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092

The above change ensures Kafka advertises its listener on localhost, which allows our local Go application to connect seamlessly. Now, you can start the Kafka broker via the following Docker command:

docker-compose up -d

Next, you’ll need to create a few directories to organize the project files. The cmd/producer and cmd/consumer directories will contain the main application files, and the pkg/models directory will store the model declarations:

mkdir -p cmd/producer cmd/consumer pkg/models

The last step is to initialize Go modules and install external packages. You’ll be using sarama to establish a connection with the Kafka broker and gin to handle the API endpoints for the notification system:

go mod init kafka-notify
go get github.com/IBM/sarama github.com/gin-gonic/gin

How to Create the User and Notification Models

With the workspace set up, the first step is to create the User and Notification structs. Move to the pkg/models directory, then create a new file named models.go and declare these structs within it:

package models

type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
}

type Notification struct {
    From    User   `json:"from"`
    To      User   `json:"to"`
    Message string `json:"message"`
}

How to Set Up the Kafka Producer 📤

The next step is to write the code for the producer. You’ll create a simple Gin web API where a user can send a notification to another user via an HTTP POST request. This request will then “produce” (send) a message to a Kafka topic named "notifications".

Let’s navigate to the cmd/producer directory and create a new file named producer.go. Within it, you’ll set up the producer logic:

package main

import (
    "encoding/json"
    "errors"
    "fmt"
    "log"
    "net/http"
    "strconv"

    "kafka-notify/pkg/models"

    "github.com/IBM/sarama"
    "github.com/gin-gonic/gin"
)

const (
    ProducerPort       = ":8080"
    KafkaServerAddress = "localhost:9092"
    KafkaTopic         = "notifications"
)

// ============== HELPER FUNCTIONS ==============
var ErrUserNotFoundInProducer = errors.New("user not found")

func findUserByID(id int, users []models.User) (models.User, error) {
    for _, user := range users {
        if user.ID == id {
            return user, nil
        }
    }
    return models.User{}, ErrUserNotFoundInProducer
}

func getIDFromRequest(formValue string, ctx *gin.Context) (int, error) {
    id, err := strconv.Atoi(ctx.PostForm(formValue))
    if err != nil {
        return 0, fmt.Errorf(
            "failed to parse ID from form value %s: %w", formValue, err)
    }
    return id, nil
}

// ============== KAFKA RELATED FUNCTIONS ==============
func sendKafkaMessage(producer sarama.SyncProducer,
    users []models.User, ctx *gin.Context, fromID, toID int) error {
    message := ctx.PostForm("message")

    fromUser, err := findUserByID(fromID, users)
    if err != nil {
        return err
    }

    toUser, err := findUserByID(toID, users)
    if err != nil {
        return err
    }

    notification := models.Notification{
        From: fromUser,
        To:   toUser, Message: message,
    }

    notificationJSON, err := json.Marshal(notification)
    if err != nil {
        return fmt.Errorf("failed to marshal notification: %w", err)
    }

    msg := &sarama.ProducerMessage{
        Topic: KafkaTopic,
        Key:   sarama.StringEncoder(strconv.Itoa(toUser.ID)),
        Value: sarama.StringEncoder(notificationJSON),
    }

    _, _, err = producer.SendMessage(msg)
    return err
}

func sendMessageHandler(producer sarama.SyncProducer,
    users []models.User) gin.HandlerFunc {
    return func(ctx *gin.Context) {
        fromID, err := getIDFromRequest("fromID", ctx)
        if err != nil {
            ctx.JSON(http.StatusBadRequest, gin.H{"message": err.Error()})
            return
        }

        toID, err := getIDFromRequest("toID", ctx)
        if err != nil {
            ctx.JSON(http.StatusBadRequest, gin.H{"message": err.Error()})
            return
        }

        err = sendKafkaMessage(producer, users, ctx, fromID, toID)
        if errors.Is(err, ErrUserNotFoundInProducer) {
            ctx.JSON(http.StatusNotFound, gin.H{"message": "User not found"})
            return
        }
        if err != nil {
            ctx.JSON(http.StatusInternalServerError, gin.H{
                "message": err.Error(),
            })
            return
        }

        ctx.JSON(http.StatusOK, gin.H{
            "message": "Notification sent successfully!",
        })
    }
}

func setupProducer() (sarama.SyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    producer, err := sarama.NewSyncProducer([]string{KafkaServerAddress},
        config)
    if err != nil {
        return nil, fmt.Errorf("failed to setup producer: %w", err)
    }
    return producer, nil
}

func main() {
    users := []models.User{
        {ID: 1, Name: "Emma"},
        {ID: 2, Name: "Bruno"},
        {ID: 3, Name: "Rick"},
        {ID: 4, Name: "Lena"},
    }

    producer, err := setupProducer()
    if err != nil {
        log.Fatalf("failed to initialize producer: %v", err)
    }
    defer producer.Close()

    gin.SetMode(gin.ReleaseMode)
    router := gin.Default()
    router.POST("/send", sendMessageHandler(producer, users))

    fmt.Printf("Kafka PRODUCER 📨 started at http://localhost%s\n",
        ProducerPort)

    if err := router.Run(ProducerPort); err != nil {
        log.Printf("failed to run the server: %v", err)
    }
}

Let’s break down the Kafka-related components within producer.go:

Within the **setupProducer()** function:

• config := sarama.NewConfig(): Initializes a new default configuration for Kafka. Think of it as setting up the parameters before connecting to the broker.
• config.Producer.Return.Successes = true: Ensures that the producer receives an acknowledgment once the message is successfully stored in the "notifications" topic.
• producer, err := sarama.NewSyncProducer(…): Initializes a synchronous Kafka producer that connects to the Kafka broker running at localhost:9092.

Inside the **sendKafkaMessage()** function:

• This function starts by retrieving the message from the context and then attempts to find both the sender and the recipient using their IDs.
• notification := models.Notification{…}: Initializes a Notification struct that encapsulates information about the sender, the recipient, and the actual message.
• msg := &sarama.ProducerMessage{…}: Constructs a ProducerMessage for the "notifications" topic, setting the recipient’s ID as the Key and the message content, which is the serialized form of the Notification as the Value.
• producer.SendMessage(msg): Sends the constructed message to the "notifications" topic.

In the **sendMessageHandler()** function:

• This function serves as an endpoint handler for the /send POST request. It processes the incoming request to ensure valid sender and recipient IDs are provided.
• After fetching the IDs, it invokes the sendKafkaMessage() function to send the Kafka message. Depending on the result, it dispatches appropriate HTTP responses: a 404 Not Found for nonexistent users, a 400 Bad Request for invalid IDs, and a 500 Internal Server Error for other failures, along with a specific error message.

Finally, within the **main()** function:

• You initialize a Kafka producer via the setupProducer() function.
• Then, you create a Gin router via gin.Default(), setting up a web server. Next, you define a POST endpoint /send to handle notifications. This endpoint expects the sender and recipient’s IDs and the message content.
• The notification is processed upon receiving a POST request via the sendMessageHandler() function, and an appropriate HTTP response is dispatched.

The above setup provides a simple way to simulate users sending notifications to each other and showcases how these notifications are produced to the "notifications" topic.

How to Set Up the Kafka Consumer 📥

After creating the producer, the next step is to set up a consumer that listens to the "notifications" topic and provides an endpoint to list notifications for a specific user.

Let’s move to the cmd/consumer directory and create a new file named consumer.go. Within it, you’ll set up the consumer logic and the Gin-based API:

package main

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "log"
    "net/http"
    "sync"

    "kafka-notify/pkg/models"

    "github.com/IBM/sarama"
    "github.com/gin-gonic/gin"
)

const (
    ConsumerGroup      = "notifications-group"
    ConsumerTopic      = "notifications"
    ConsumerPort       = ":8081"
    KafkaServerAddress = "localhost:9092"
)

// ============== HELPER FUNCTIONS ==============
var ErrNoMessagesFound = errors.New("no messages found")

func getUserIDFromRequest(ctx *gin.Context) (string, error) {
    userID := ctx.Param("userID")
    if userID == "" {
        return "", ErrNoMessagesFound
    }
    return userID, nil
}

// ====== NOTIFICATION STORAGE ======
type UserNotifications map[string][]models.Notification

type NotificationStore struct {
    data UserNotifications
    mu   sync.RWMutex
}

func (ns *NotificationStore) Add(userID string,
    notification models.Notification) {
    ns.mu.Lock()
    defer ns.mu.Unlock()
    ns.data[userID] = append(ns.data[userID], notification)
}

func (ns *NotificationStore) Get(userID string) []models.Notification {
    ns.mu.RLock()
    defer ns.mu.RUnlock()
    return ns.data[userID]
}

// ============== KAFKA RELATED FUNCTIONS ==============
type Consumer struct {
    store *NotificationStore
}

func (*Consumer) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (*Consumer) Cleanup(sarama.ConsumerGroupSession) error { return nil }

func (consumer *Consumer) ConsumeClaim(
    sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        userID := string(msg.Key)
        var notification models.Notification
        err := json.Unmarshal(msg.Value, &notification)
        if err != nil {
            log.Printf("failed to unmarshal notification: %v", err)
            continue
        }
        consumer.store.Add(userID, notification)
        sess.MarkMessage(msg, "")
    }
    return nil
}

func initializeConsumerGroup() (sarama.ConsumerGroup, error) {
    config := sarama.NewConfig()

    consumerGroup, err := sarama.NewConsumerGroup(
        []string{KafkaServerAddress}, ConsumerGroup, config)
    if err != nil {
        return nil, fmt.Errorf("failed to initialize consumer group: %w", err)
    }

    return consumerGroup, nil
}

func setupConsumerGroup(ctx context.Context, store *NotificationStore) {
    consumerGroup, err := initializeConsumerGroup()
    if err != nil {
        log.Printf("initialization error: %v", err)
    }
    defer consumerGroup.Close()

    consumer := &Consumer{
        store: store,
    }

    for {
        err = consumerGroup.Consume(ctx, []string{ConsumerTopic}, consumer)
        if err != nil {
            log.Printf("error from consumer: %v", err)
        }
        if ctx.Err() != nil {
            return
        }
    }
}

func handleNotifications(ctx *gin.Context, store *NotificationStore) {
    userID, err := getUserIDFromRequest(ctx)
    if err != nil {
        ctx.JSON(http.StatusNotFound, gin.H{"message": err.Error()})
        return
    }

    notes := store.Get(userID)
    if len(notes) == 0 {
        ctx.JSON(http.StatusOK,
            gin.H{
                "message":       "No notifications found for user",
                "notifications": []models.Notification{},
            })
        return
    }

    ctx.JSON(http.StatusOK, gin.H{"notifications": notes})
}

func main() {
    store := &NotificationStore{
        data: make(UserNotifications),
    }

    ctx, cancel := context.WithCancel(context.Background())
    go setupConsumerGroup(ctx, store)
    defer cancel()

    gin.SetMode(gin.ReleaseMode)
    router := gin.Default()
    router.GET("/notifications/:userID", func(ctx *gin.Context) {
        handleNotifications(ctx, store)
    })

    fmt.Printf("Kafka CONSUMER (Group: %s) 👥📥 "+
        "started at http://localhost%s\n", ConsumerGroup, ConsumerPort)

    if err := router.Run(ConsumerPort); err != nil {
        log.Printf("failed to run the server: %v", err)
    }
}

Let’s examine the Kafka-related operations within consumer.go:

Within the **initializeConsumerGroup()** function:

• config := sarama.NewConfig(): Initializes a new default configuration for Kafka.
• consumerGroup, err := sarama.NewConsumerGroup(…): Creates a new Kafka consumer group that connects to the broker running on localhost:9092. The group name is "notifications-group".

Inside the **Consumer** struct and its methods:

• The Consumer struct has a store field, which is a reference to the NotificationStore to keep track of the received notifications.
• Setup() and Cleanup() methods are required to satisfy the sarama.ConsumerGroupHandler interface. While they will NOT be used in this tutorial, they can serve potential roles for initialization and cleanup during message consumption but act as placeholders here.
• In the ConsumeClaim() method: The consumer listens for new messages on the topic. For each message, it fetches the userID (the Key of the message), un-marshals the message into a Notification struct, and adds the notification to the NotificationStore.

In the **setupConsumerGroup()** function:

• This function sets up the Kafka consumer group, listens for incoming messages, and processes them using the Consumer struct methods.
• It runs a for loop indefinitely, consuming messages from the “notifications” topic and processing any errors that arise.

The **handleNotifications()** function:

• Initially, it attempts to retrieve the userID from the request. If it doesn’t exist, it returns a 404 Not Found status.
• Then, it fetches the notifications for the provided user ID from the NotificationStore. Depending on whether the user has notifications, it responds with a 200 OK status and either an empty notifications slice or sends back the current notifications.

Lastly, within the **main()** function:

• store := &NotificationStore{…}: Creates an instance of NotificationStore to hold the notifications
• ctx, cancel := context.WithCancel(context.Background()): Sets up a cancellable context that can be used to stop the consumer group.
• go setupConsumerGroup(ctx, store): Starts the consumer group in a separate Goroutine, allowing it to operate concurrently without blocking the main thread.
• The last step is to create a Gin router and define a GET endpoint /notifications/:userID that will fetch the notifications for a specific user via the handleNotifications() function when accessed.

This setup provides a straightforward way to consume messages from the "notifications" topic and present them to users through a web endpoint.

Let's Test the Real-Time Notification System👨‍🔬🖥️👩‍🔬

Now that the producer and consumer are ready, it’s time to see the system in action.

1. Start the producer

Open up a terminal, move into the kafka-notify directory, and run the producer with the following command:

go run cmd/producer/producer.go

2. Start the consumer

Open up a second terminal window, navigate to the kafka-notify directory, and start the consumer by running:

go run cmd/consumer/consumer.go

3. Sending notifications

With both producer and consumer running, you can simulate sending notifications. Open up a third terminal and use the below curl commands to send notifications:

User 1 (Emma) receives a notification from User 2 (Bruno):

curl -X POST http://localhost:8080/send \
-d "fromID=2&toID=1&message=Bruno started following you."

User 2 (Bruno) receives a notification from User 1 (Emma):

curl -X POST http://localhost:8080/send \
-d "fromID=1&toID=2&message=Emma mentioned you in a comment: 'Great seeing you yesterday, @Bruno!'"

User 1 (Emma) receives a notification from User 4 (Lena):

curl -X POST http://localhost:8080/send \
-d "fromID=4&toID=1&message=Lena liked your post: 'My weekend getaway!'"

4. Retrieving notifications

Finally, you can fetch the notifications of a specific user. You can use the below curl commands to fetch notifications:

Retrieving notifications for User 1 (Emma):

curl http://localhost:8081/notifications/1

Output:

{"notifications": [{"from": {"id": 2, "name": "Bruno"}, "to": {"id": 1, "name": "Emma"}, "message": "Bruno started following you."}]}
{"notifications": [{"from": {"id": 4, "name": "Lena"}, "to": {"id": 1, "name": "Emma"}, "message": "Lena liked your post: 'My weekend getaway!'"}]}

In the above output, you see a JSON response listing all the notifications for Emma. As you send more notifications, they accumulate, and you can fetch them using the consumer’s API.

Wrapping up 📝

In this tutorial, you’ve learned how to set up a basic real-time notification system using Kafka in Go.

By simulating the process of users sending and fetching notifications, you’ve gained hands-on experience with Kafka components. This is a foundational step in understanding how Kafka can be integrated into Go applications for various real-time data processing tasks.

You can access the entire codebase for this project in the following GitHub repository: https://github.com/gutyoh/kafka-notify

If you found this article helpful and want to expand your knowledge on Golang 🐿️, Docker 🐳, and Gin 🍸, consider checking out the Introduction to Go track, which covers basic Golang concepts.

Registered users can also check the Go Developer and Introduction to Docker tracks on Hyperskill. I've been actively involved as an expert in curating these tracks, ensuring they deliver top-notch educational content.

Before concluding, I owe a big thank you to my dear friend Anton Illarionov. His expertise in integrating Go with Kafka inspired the idea for this article. You can explore his projects on GitHub.

Let me know if you have any questions or feedback regarding this article.

Thank you for reading, and keep on coding!

Hi, friends, and welcome to another tutorial. We'll be talking about RabbitMQ, a tool that is widely used as a message broker in distributed and Pub-Sub systems.

While multiple applications can share a single RabbitMQ instance, it's a great approach to configure it to have different virtual hosts for each application.

One practical usage in a Python application is using it as a Celery broker URL or simply as a message queue.

One benefit of this is that it saves the costs that comes with spinning up multiple instances of RabbitMQ. Creating RabbitMQ virtual hosts is similar both locally and when running in a self-managed solution such as AWS RabbitMQ

By the end of this guide, you will be able to:

1. Create multiple virtual hosts for a single RabbitMQ instance.
2. Test the connections using the correct credentials, that is user, password, and virtual host.

Prerequisites to follow along:

1. Ensure you have Docker and Docker Compose installed, or
2. Have a self-managed RabbitMQ instance from AWS, CloudAMQP, or any other service provider

Without any delay, let’s get started

How to Run a RabbitMQ Instance Locally

To keep things simple, we are going to run the RabbitMQ instance in a Docker container.

Create a file named docker-compose.yml and update it with the following content:

docker-compose.yml

version: "3.8"
services:
  rabbitmq:
    image: rabbitmq:3.8-management-alpine
    container_name: 'rabbitmq_test'
    environment:
      - RABBITMQ_DEFAULT_USER=sampleuser
      - RABBITMQ_DEFAULT_PASS=samplepassword
    ports:
        - 5672:5672
        - 15672:15672
    volumes:
      - ~/.docker-conf/rabbitmq/data/:/var/lib/rabbitmq/
      - ~/.docker-conf/rabbitmq/log/:/var/log/rabbitmq

To start the container, run this command in the terminal:

docker-compose up

Once the container is up and running, access the GUI by visiting this link: http://localhost:15672/

RabbitMQ GUI login page

Use the credentials specified in the docker-compose file to gain access to the dashboard – “sampleuser” and “samplepassword” in my case.

Note: If you are using a self-managed RabbitMQ instance from AWS or CloudAMQP, the RabbitMQ connection string will look similar to this:

amqps://ausername:apassword@32wewjijiokkoo.mq.eu-west-3.amazonaws.com:5671

To access the RabbitMQ admin dashboard, follow these steps:

1. Visit the URL using the HTTPS protocol: https://32wewjijiokkoo.mq.eu-west-3.amazonaws.com
2. Remove the username, password, and port from the URL.
3. After accessing the URL, the system will prompt you to enter the username and password specified in the RabbitMQ connection string (“ausername” and “apassword” in this case). Provide these credentials to log in to the admin dashboard.

How to Create a Virtual Host

Once you are logged in, click on the ‘Admin’ tab, and then navigate to the ‘Virtual Hosts’ tab as shown in the image below.

RabbitMQ Virtualhost tab

Click on ‘Add a new virtual host’, and provide it with an appropriate name. Finally, click on the ‘Add virtual host’ button to create the new virtual host.

New Virtual Host created

How to Create a New User

On the Users tab, click on the ‘Add a user’ toggle. Specify a Username and Password for the new user. If you wish, you may grant the user access to the Admin (GUI) interface. Finally, click on the ‘Add user’ button to create the new user.

Create a new user

How to Assign the New User to the Created Virtual Host (Vhost)

On the Users tab, locate and click on the previously created user, which was named ‘user1’ in my case.

New user in users list

For the new user, choose the virtual host as ‘new-virtual-host’, and then click the ‘Set permission’ button. This grants the user access to the specified virtual host. To confirm this, go back to the Users tab and verify that the user now appears with access to the ‘new-virtual-host'

Assign new user to the created virtual host

By now, the new user 'user1' has been configured to access the vhost 'new-virtual-host'.

New user assigned to the created virtual host

How to Test the Connection for the New User and Vhost

To test if the new user can access the new vhost, you can use the following sample Python script. Once everything is working correctly, you can use the URL string to connect to the RabbitMQ broker in any application.

_testrabbit.py

import pika

def test_rabbitmq_url(url):

    try:
        params = pika.URLParameters(url)
        connection = pika.BlockingConnection(params)
        connection.close()
        print("Connection successful!")

    except pika.exceptions.AMQPError as e:
        print("Connection failed:", e)

test_rabbitmq_url('amqp://user1:password@localhost:5672/new-virtual-host')

Note how we have specified the RabbitMQ URL: it has the username as “user1” and password as “password9,” while the virtual host is named ‘new-virtual-host.’

Remember to install pika by running pip install pika on the command line.

To run this script on the terminal, enter the following command:

python test_rabbit.py

Successfully connected to RabbitMQ

This indicates a successful connection to the RabbitMQ instance.

Change the password in the utility function to an incorrect one, and you should see an outcome similar to this:

Connection to RabbitMQ failed

Wrapping Up

In conclusion, creating multiple virtual hosts on a single RabbitMQ instance can significantly enhance your message queue management capabilities.

By effectively segregating applications and resources, you ensure better organization, security, and scalability. With the step-by-step guide provided in this article, you now have the knowledge to harness the full potential of RabbitMQ

Thanks for reading, and I hope you enjoyed the article.

For more interesting content and tips, please feel free to check out my video collection on YouTube and consider hitting the subscribe button.

You can also find me on LinkedIn.

There are a number of tools in this field, but RabbitMQ and Apache Kafka are two of the most popular. While both are robust and reliable, they have unique features and use cases that make them distinct.

RabbitMQ is an open-source message broker. It's recognized for its flexibility and support for various messaging protocols. Apache Kafka is rapidly gaining popularity and is known for its ability to handle real-time data feeds with low latency.

In this tutorial, I'll focus on RabbitMQ, its core features, and how you can use it to effectively build scalable, loosely coupled applications.

Stay with me as we explore the world of RabbitMQ, its unique capabilities, and how it sets itself apart in the ever-evolving landscape of message queue technologies.

What is RabbitMQ?

RabbitMQ is an open-source message broker software (also called a message-oriented middleware) that implements the Advanced Message Queuing Protocol (AMQP). It provides a common platform for sending and receiving messages.

RabbitMQ supports multiple messaging protocols and can be deployed in distributed and federated configurations to meet high-scale, high-availability requirements.

When Should You Use RabbitMQ?

Consider an e-commerce website (like Amazon) where users can place orders that need to be processed.

The order processing system might involve several steps, such as inventory checks, payment processing, shipping, and so on, each of which can potentially take some time and are ideally handled asynchronously.

1. Inventory Checks: When a user places an order, the system should check whether the ordered products are in stock. You can send a message to a queue that is consumed by a service responsible for checking inventory. This way, even if the inventory service is temporarily down or overloaded, the order messages won't be lost – they'll be processed as soon as the service is available.
2. Payment Processing: The payment processing might be done by another microservice. Once the inventory check is completed, a message can be sent to a queue for the payment service. This decouples the payment processing from the inventory checking, allowing these operations to scale independently.
3. Shipping: After payment confirmation, a message can be sent to another queue that's responsible for handling shipping. Again, this service might take some time, but because it's decoupled from the rest of the system, it won't slow down other operations.
4. Notification system: After every successful order placement, payment, and shipment, notifications (Email or SMS) must be sent to the customer. This can be handled by separate services that listen to specific queues.

How to Install and Run RabbitMQ

In this example, I'll be showing you how to use Docker to run RabbitMQ. But if you prefer, you can install and run it manually on your system. The official documentation provides a detailed guide on how to do this.

I find Docker to be a convenient tool for running RabbitMQ because it simplifies the setup and management processes. If you're new to Docker, I recommend reading my previous tutorials related to Docker for a thorough understanding.

To get started, you'll need to pull the Docker image from Docker Hub.

docker pull rabbitmq

Official Docker image of RabbitMQ in Docker Desktop

Before running the image, you need to map the two port numbers (15672 and 5672).

1. Port 5672: This is the default port for RabbitMQ when using AMQP (Advanced Message Queuing Protocol). Clients that connect with AMQP typically use this port. So, if you're using an AMQP client library to connect to RabbitMQ, it's likely to connect on port 5672.
2. Port 15672: This is the default port for the RabbitMQ Management UI, when using the rabbitmq_management plugin. The Management UI is a web-based interface that allows you to monitor and control your RabbitMQ server.

I'm running RabbitMQ on the Docker by adding the above-mentioned ports. Refer the below screenshot for additional reference.

Port configuration to run the RabbitMQ Image.

Our RabbitMQ Server is up and running in Docker.

RabbitMQ Server container running in Docker

How to Implement a Message Queue in NodeJS

I'll using amqplib, which is a popular NodeJS library that provides an API for interacting with RabbitMQ. It supports all the features of RabbitMQ's AMQP 0-9-1 model, including things like confirm channels, exchanges, queues, bindings, and message properties.

I have been using the term AMQ Protocol in this tutorial and I feel like this is the right time to give a quick introduction to it.

What is AMQP?

AMQP stands for Advanced Message Queuing Protocol. It is an open-standard protocol for message-oriented middleware. The defining features of AMQP are message orientation, queuing, routing (including point-to-point and publish-and-subscribe), reliability, and security.

AMQP has the following components:

1. Producer is an application that sends messages.
2. Consumer is an application that receives messages.
3. Queue is a buffer that stores the messages.
4. Message is the information that is sent from the producer to a consumer.
5. Exchange receives messages from producers and pushes them to queues depending on rules defined by the exchange type. The exchange type determines how messages are routed.
6. Binding links the queue to the exchange.

The AMQP protocol enables standardized communication between different applications, making it a good choice for a messaging system in a microservices architecture. This protocol can ensure that a message is delivered not just to the messaging system, but all the way to the correct consumer.

You'll remember the example that I described at the beginning of this article about the high level implementation of message queues in a E-commerce site.

Let's go through the same but a bit deeper in the RabbitMQ context.

Order Placement

When a customer places an order on the e-commerce website, the order service produces a message to a RabbitMQ exchange. The message contains information about the product ID and the quantity ordered.

Inventory Update

An inventory service is set up as a consumer to receive messages from a queue bound to the exchange. Once it receives a message, it reduces the inventory for the specified product by the ordered quantity. If the inventory is insufficient, it can send a message back to the order service to indicate the problem.

Order Confirmation

Once the inventory service successfully updates the inventory, it'll send a message to the order service. The order service, set up as a consumer for this exchange, can then update the order status and notify the customer.

Benefits

Resilience

Let's assume your inventory service is down for some time. Then the messages in the RabbitMQ queue will stay there and won't be lost. Once the inventory service is back online, it'll continue processing the messages from where it left off.

Scalability

During high traffic periods, more instances of the inventory service can be launched, all consuming messages from the same queue. This enables load balancing and ensures that the system can handle the increased load.

Let's come back to our implementation. In this example, we'll send a message from sender to our receiver. On the receiver end, we print the message on the console.

How to Create the Sender (Producer)

This is the component or part of our application that creates and sends messages to the messaging queue.

The sender does not send messages directly to the consumer. Instead, it sends the messages to an exchange in RabbitMQ. The exchange then routes the messages to the appropriate queue based on certain criteria.

Here we're creating a queue called product_inventory. Alternatively, you can clone my repo from here.

We can send only byte arrays in the message, So I convert the message object to a string and send it to the queue. From the above code, you can understand that we're creating a channel and sending a message through it.

How to Create the Receiver (Consumer)

This is the component or part of our application that receives and processes the messages from the queue. A consumer can continuously poll the queue for new messages or be set up to automatically trigger when a new message is added to the queue.

Here we're listening for messages. Alternatively, you can clone my repo from here.

In the above code, we're listening for messages (consume) and print them on the console once we receive it.

In the context of our specific use case, the file containing the message sender (or producer) should typically be located in the root directory of our e-commerce site project. This is where we generate and send messages based on user actions, such as placing an order.

On the other hand, the file containing the message receiver (or consumer) should ideally be located in the inventory management service. This is because the inventory management service is responsible for processing these messages, such as updating the inventory when an order is placed.

Let's run our receiver service first:

yarn start

The initial output of the consumer service looks like this:

Initial output in the consumer service - Listening for messages

Once we run our sender, a message will be sent to the consumer. Run the same command yarn start on the sender repo.

Here's the output of it:

Sending a message from the producer to the consumer

Hurray! We received a message sent from the RabbitMQ producer in our consumer service.

Received message from the RabbitMQ producer in our consumer service

Conclusion

In this article, we've explored the basics of RabbitMQ, a robust and efficient message broker, and demonstrated its application in a NodeJS environment.

Using a simple e-commerce scenario, we showcased how to set up a sender (producer) and a consumer to handle asynchronous messages between different components of our application. But in real-world applications, you will likely encounter more complex scenarios that require advanced integrations and the usage of RabbitMQ.

To navigate these complexities, it's crucial to have a solid understanding of RabbitMQ's underlying concepts and its AMQP protocol. As you delve deeper into RabbitMQ, you'll find it to be an incredibly versatile tool, capable of handling a wide range of messaging needs, and ultimately helping you build scalable, decoupled, and resilient applications.

Check out the source code of the project on GitHub: rabbit-sender, rabbit-receiver.

To learn more about RabbitMQ, subscribe to my email newsletter on my site (https://5minslearn.gogosoon.com) and follow me on social media.

An introduction to RabbitMQ, message broker, AMQP model and more.

In distributed systems, communication between various applications plays an important role. Effectively passing messages between applications was always a crucial decision in system design. One of the elegant solutions to pass messages around your distributed system is a message broker.

In distributed systems, communication between various applications plays an important role. Effectively passing messages between applications was always a crucial decision in system design. One of the elegant solutions to pass messages around your distributed system is a message broker.

They bring in decoupling between the applications and provide an effective way to communicate. With message brokers, an application needs no prior knowledge of their recipients to communicate.

However, what is RabbtiMQ? How does RabbitMQ fit in this picture? Also, what is AMQP?

By the end of this article, we will be able to answer these questions. I also added a few animations so that you can visualize RabbitMQ concepts.

So are you excited? I am! If you ever had difficulties understanding message brokers, as I did, then this article is the right place to start your journey. Stay with me ?

Message Broker

In general, a broker is a person who facilitates trades between a buyer and a seller. An example could be a real estate agent or a stockbroker.

Similarly, if we want to trade messages between two distributed software components, we need a mediator. This mediator is known as the message broker. It receives incoming messages from a sender and sends them to a recipient. This way the sender and receiver can be totally isolated.

Another analogy for a message broker can be a Post Office (see Figure 1). Let’s take a scenario where you are going to send a letter to your cousin living in another city. Then as per this analogy, you are a producer, your cousin is a consumer, and the post office is a message broker.

Figure 1: Analogy for message broker

RabbitMQ as message broker

Now we know that the purpose of a message broker is to route messages from a producer to a consumer. Let’s examine one such message broker — RabbitMQ. It’s one of the most extensively used message brokers these days.

The way RabbitMQ routes messages depends upon the messaging protocol it implements. RabbitMQ supports multiple messaging protocols. However, the one we are interested in is AMQP. It is an acronym for Advanced Message Queuing Protocol.

So without any further ado let’s have a closer look at the AMQP protocol model.

Advanced Message Queuing Protocol

The conceptual model of AMQP is quite simple and straightforward. It has three entities:

1. Queue
2. Binding
3. Exchange

When a publisher pushes a message to RabbitMQ, it first arrives at an exchange. The exchange then distributes copies of these messages to variously connected queues. Finally, consumers receive these messages.

Consider a message as a piece of data. It is necessarily a package with a payload and some meta-data. The payload contains full data whereas meta-data are properties used by RabbitMQ.

Figure 2 depicts a graphical representation of the AMQP model.

Figure 2: AMQP model

AMQP is a programmable protocol. Programmers have the liberty to use libraries to configure entities (exchange, binding, and queue) as per their own needs. A RabbitMQ admin has no role in setting up these entities.

There are plenty of libraries available to work with RabbitMQ. You can choose from Nodejs, Python, .Net, Java, and many more.

Queues

These queues are somehow similar to the queues from our data structure classes. RabbitMQ queues also follow FIFO — First-In-First-Out methodology. A queue is a place where RabbitMQ stores messages/data.

Programmers can configure queues through available programming libraries. You can make a queue durable ( with the Durability property) to safeguard your data in case the broker crashes. You can also provide a name(with the Name property) to a queue. Other than Name and Durability, a queue has a few other properties like auto-delete, exclusive, and arguments.

Before moving any further, it’s important to understand who is a direct consumer of these queues. Moreover, how many ways can a user consume messages from a queue?

Consumers

Consumers are the ones who are going to use messages stored in a queue. It is possible to connect more than one consumer to a queue at a time. Consumers can either pull the message from the queue by pooling it or queues can even push the message to various connect consumers.

Bindings

Bindings are the rules that a queue defines while establishing a connection with an exchange. You can have a queue connected to multiple exchanges. Every queue is also connected to a default exchange. An exchange will use these bindings to route messages to queues.

Exchanges and their types

An Exchange is a gateway to RabbitMQ for your messages. The distance the message has to travel inside RabbitMQ depends on the type of exchange. Primarily there are four types.

• Direct
• Fanout
• Topic
• Header

Direct

The name explains it all! — A direct exchange delivers a message directly to the queues that satisfy the below condition:

Routing key == Binding key

A routing key is an attribute of the message. On the other hand, a binding key is something you specify while creating a binding between a queue and an exchange.

Figure 3 is a visual explanation of how messages flow while using a direct exchange.

A message originates from a producer (green circle) with a routing key — img.resize. Once it reaches the exchange (Orange circle), the exchange will try to find all queues with binding key — img.resize. In case of a match, the message is pushed to all matched queues (resize in our case). If there is no match found, the message can be sent back to the producer or can even be discarded. We are lucky that we found a match in our example ?

[gif image]

Once the message reaches the desired queue (resize in our case), they are distributed in round-robin fashion to all the connected consumers (resizer.1/resizer.2 in our case).

By distributing messages in a round-robin fashion, RabbitMQ makes sure that the messages are load balanced.

You must have noticed that the queue named crop is not receiving any messages. Because the routing key in this example is img.resize. To send messages to this queue, we need to send messages with a routing key that would match the binding key (say img.crop for instance).

Fanout

A Fanout exchange ignores routing keys and distributes a message to all the connected queues. No wonder it is called Fanout (blowing messages to all connected queues! ?).

One of the use cases for this type of exchange is message broadcast.

Please note that RabbitMQ will still do round robin if there is more than one consumer of the queue.

Topic

A topic exchange routes a message by matching routing key with a pattern in the binding key.

Routing key == Pattern in binding key.

RabbitMQ uses two wild card characters for pattern matching * and #. Use a * to match 1 word and a # to match 0 or more words.

Figure 5 is a visual depiction of a topic exchange. Messages with routing key — logs.error will match patterns — logs.error and logs.*. Hence these messages will end up in the queues — only error and all logs.

Whereas for the producer at the bottom-left, messages with routing key— logs.success will match patterns of binding key #success and logs.* . Hence these messages will end up in the queues — all logsand only success .

[gif]

This type of exchange has a vast range of use cases. It can be used in the publish-subscribe pattern, distributing relevant data to desiring workers processes and many more.

Header

A header is a particular type of exchange that routes messages based on keys present in the message header. It overlooks routing key attribute of the message.

When creating bindings for a header exchange, it is possible to bind a queue to match more than one header. In such a case, RabbitMQ should know from the producer if it should match all or any of these keys.

A producer/application can do this by providing an extra flag called ‘x-match’. ‘x-match’ can have any or all values. The first one mandates that only one value should match while the latter mandates that all must match.

Message Acknowledgement

Once a message reaches its destination, the broker should delete the message from the queue. It is necessary because a queue overflow can occur if it keeps accumulating messages.

Before deleting any message, the broker must have a delivery acknowledgment. There are two possible ways to acknowledge message delivery.

1. Automatic acknowledgment: Once a consumer receives the message
2. Explicit acknowledgment: When a consumer sends back an acknowledgment

In most cases, explicit acknowledgment is used as it makes sure that the consumer has consumed the message without any failover.

What’s next

RabbitMQ is a very mature and useful product. This article is only a high-level introduction to RabbitMQ. I simplified the concepts to provide a reference point for you to move further. Visit the RabbitMQ website for more complex topics.

Hope you like the article. Don’t forget to clap(or applaud ?). Follow to read my upcoming stories. Until next time, keep Queuing.