Produce And Consume Messages To RabbitMQ Docker Container Using .NET Client

Introduction

In this article, I will showcase how to connect, produce, and consume messages to a RabbitMQ instance running as a docker container through the C# .NET client.

As a prerequisite, I assume you are familiar with C#, Docker, and RabbitMQ.

This article covers pulling a RabbitMQ image from the docker hub and then containerizing it; creating a dummy .NET Core Web API that produces messages into the RabbitMQ queue; and developing a dummy consumer console application that listens to the queue to consume messages.

The idea here is not to dwell too much into the intricacies and configurations of the RabbitMQ jargon, but to illustrate the way to connect to an instance running in a container through a dummy producer and consumer.

So let's get started!

RabbitMQ Docker Image and Container Setup

As a first step, we should have the Docker Desktop downloaded and installed in our system.

After that, we would be running the following command to pull the desired RabbitMQ image from the respective DockerHub page.

docker pull rabbitmq:3-management

This will pull the RabbitMQ image from the docker hub with a tag of "3-management", which includes the management plug-in with this. Once the image is downloaded successfully, we can run that and put it into a container with the below command.

docker run -d --hostname my-rabbit --name my-rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management

Let's now understand segment-by-segment what this script does.

This script creates a container with the hostname "my-rabbit" and performs port mapping. This port mapping is a very important step because the underlying support of ports for the AMQP client to connect to the RabbitMQ instance needs to be compatible with our desired port in the host.

Here, the TCP container ports of 5672 and 15672 are mapped to the same ports in the host. As per the RabbitMQ documentation "5672, 5671: used by AMQP 0-9-1 and 1.0 clients without and with TLS". Our RabbitMQ instances should expose the supported AMQP port and our Producers and Consumers should be connected to that port for successful communication.

After successful execution of the above docker command, on "docker ps", we should be able to see the following container running.

Next, let's set up our dummy Producer!

The Producer

Here, I have created an ASP.NET Core WebAPI project to simulate a producer. I have added a custom controller named: ConfigurationController which would expose a POST API for producing messages. First, let's see how we can connect to the running RabbitMQ container.

We would be using the NuGet package of the .NET RabbitMQ client: RabbitMQ.Client (at the time of writing this article, I had installed version 6.0.0).

Connecting to the Container

There are two ways we could initialize the ConnectionFactory to instantiate connection with the running RabbitMQ container.

Step 1. Using the host and port information along with credentials. For simplicity, I have not changed the default credential - which is set to 'guest' for both username and password.

var factory = new ConnectionFactory()
{
    HostName = "localhost",
    UserName = ConnectionFactory.DefaultUser,
    Password = ConnectionFactory.DefaultPass,
    Port = AmqpTcpEndpoint.UseDefaultPort
};

Step 2. Alternatively, we could also set the "Uri" property in the factory instance like this.

var factory = new ConnectionFactory()
{
    Uri = new Uri("amqp://guest:guest@localhost:5672")
};

Although passing the credential in the Uri is not necessary since I am using the default credential, here I have kept it in the above example for illustration purposes.

Declare a Queue and Publish a Message

The below code snippet shows how we can create a connection with the factory, declare a queue, and then publish a message to the RabbitMQ container.

CONGIG_QUEUE is a class-level const for my queue name.

private const string CONFIG_QUEUE = "configQueue";

using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queue: CONFIG_QUEUE,
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    string message = "CPUThreshold: " + config.CPUThreshold;
    var body = Encoding.UTF8.GetBytes(message);

    channel.BasicPublish(exchange: "",
                         routingKey: CONFIG_QUEUE,
                         basicProperties: null,
                         body: body);
}

As you can see, I am using an imaginary CPUThreshold value that could be passed to the POST method in the controller through an API call which I encode into a message and publish through a queue.

With this, we complete our implementation of the producer demo.

Next up, let's create a dummy Consumer and consume the message.

The Consumer

I have created the consumer here as a console application. As with the producer, here we start with establishing a connection to the running container queue. We must connect to the same queue as the producer to listen to the messages published. After that, we consume the message produced and send an "Ack" back. The message is removed from the queue.

The below code snippet is self-explanatory.

// establish connection
var factory = new ConnectionFactory()
{
    Uri = new Uri("amqp://guest:guest@localhost:5672")
};
var rabbitMqConnection = factory.CreateConnection();
var rabbitMqChannel = rabbitMqConnection.CreateModel();

// declare the queue
rabbitMqChannel.QueueDeclare(queue: CONFIG_QUEUE,
                             durable: false,
                             exclusive: false,
                             autoDelete: false,
                             arguments: null);

rabbitMqChannel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

// consume the message received
var consumer = new EventingBasicConsumer(rabbitMqChannel);
consumer.Received += (model, args) =>
{
    var body = args.Body;
    var message = Encoding.UTF8.GetString(body.ToArray());
    Console.WriteLine("CPUThreshold Changed To: " + message);
    rabbitMqChannel.BasicAck(deliveryTag: args.DeliveryTag, multiple: false);
    Thread.Sleep(1000);
};
rabbitMqChannel.BasicConsume(queue: CONFIG_QUEUE,
                             autoAck: false,
                             consumer: consumer);
Console.ReadLine();

Let's see all these together!

Now that it's all developed and set up, let's see them working.

RabbitMQ Management

With the container set up and configured as per our first step, we should be able to log into the management console of RabbitMQ at this address: http://localhost:15672/

We would need to log in using the default credential as 'guest' for both the username and password.

After a successful login, we should be able to see that there are no queues presently created under the "Queue" tab.

Spin up the Producer Web API & Send POST request through Postman

After we have successfully spun up our Producer Web API, we can send a POST call through Postman.

With this successful POST call, if we go back and observe the management console, we will see a new queue created under the "Queue" tab with the same name that we have specified in our code.

Then if we see the queue, we will see one message has been successfully queued and is waiting to be consumed.

Run the Consumer Console App

Now, we would run the consumer app to start listening to the queue in the RabbitMQ instance and start consuming the message. After successful execution of the console app, we should see the following output in the console window.

This confirms the successful consumption of the message from the queue.

Let's summarize everything!

So, we first started up by pulling the RabbitMQ image from the docker hub and then containerizing it and mapping it to a compatible port in the container host. After that, we ran the container and ensured it was running. We then developed dummy producer and consumer apps to establish a connection to the running container and produce and consume the messages through the open-source message broker instance hosted in the docker container.

This is the full view where I have tried to show the management console, the Postman sending POST messages to trigger the message publications, and the consumer console app output. As you can see here, with repetitive messages being published with varying CPU Threshold values, the messages are published and consumed via the message broker instance which is running as a docker container.