How to work with standard queue in RabbitMQ?

RabbitMQ‘s standard queue is a very basic way of using a queue which is known to most of the developers. Microsoft’s MSMQ also uses the same standard queue concept. In this post, I’ll explain how to work with standard queue in RabbitMQ by considering a very simple payment scenario.

Scenario

Consider an e-commerce application that needs a payment mechanism to process the payment from customers. It is possible to have this processing built-in to the same application. However, it is also possible that a third-party application should be responsible to take those payment requests and process them further. This can simplify the overall application by giving specific responsibility among applications.

Solution

This problem can be easily solved by a message queue or message broker software such as RabbitMQ. The publisher (main e-commerce app) can publish a message to the queue from where a consumer (payment processor app) can read the payment request to process it further.

Concept of Standard Queue

A publisher publishes a message to the queue which is then consumed by a consumer. At high level, it looks like that the publisher directly sends a message to the queue but behind the scenes, message is published to the default direct exchange of RabbitMQ. This exchange then uses the queue name to find a matching queue or create a new one if not available and passes that message to the queue. Below is an image referenced from RabbitMQ official website which explains a basic flow of a message visually. This image illustrates how a route such as “Hello, world” is matched by an exchange to decide a queue to select. Click on the image to read more details on RabbitMQ website.

Process Concept – How to work with standard queue in RabbitMQ?

Demo app

The source is available at Github. Please let me know if you have any issues to run the application. I’ll update this repository with other examples as well in future.

Make sure you have RabbitMQ installed on your system. I’m using Visual Studio 2015 to create this application. The console application uses RabbitMQ.Client package which is available from Nuget. I have created a windows console application to demonstrate how to create a queue, send/receive a message. To keep this simple, the same application sends and receives the messages in the console application. But in a real world, these can be two separate applications hosted in different environments, provided they connect to the message broker software i.e. RabbitMQ.

To work on a payment process, I’ve created a `Payment.cs` file that defines a `Payment` class mentioned below. This type will be used by both publisher and consumer in the process.

[Serializable]
public class Payment
{
    public decimal AmountToPay { get; set; }
    public string CardNumber { get; set; }
    public string Name { get; set; }
}

Next, you will see that the class above is decorated with `Serializable` attribute. This is required so that an object of this type can be converted to bytes to send it as a message from publisher to consumer via RabbitMQ. For the serialization and deserialization, I’ve following class helper that does this job:

public static class ObjectSerializer
{
    public static byte[] Serialize(this object obj)
    {
        if (obj == null)
        {
            return null;
        }

        var json = JsonConvert.SerializeObject(obj);
        return Encoding.ASCII.GetBytes(json);
    }

    public static object Deserialize(this byte[] arrBytes, Type type)
    {
        var json = Encoding.Default.GetString(arrBytes);
        return JsonConvert.DeserializeObject(json, type);
    }

    public static string DeserializeText(this byte[] arrBytes)
    {
        return Encoding.Default.GetString(arrBytes);
    }
}

This is what is needed to actually write code in the `Program.cs` file which is mentioned below:

class Program
{
	private static ConnectionFactory _connectionFactory;
	private static IConnection _connection;
	private static IModel _model;

	private const string QueueName = "StandardQueue";

	static void Main(string[] args)
	{
	    var payments = new List<Payment>();

	    for (int i = 1; i <= 10; i++)
	    {
	        var payment = new Payment() { AmountToPay = i * 5, CardNumber = "4657342588996543", Name = "Mr. X " };
	        payments.Add(payment);
	    }

   	    CreateQueue();

        payments.ForEach(payment => SendMessage(payment));

	    Console.WriteLine("========================================================================================================");

        Receive();

	    Console.ReadLine();
	}

	private static void Receive()
	{
	    var consumer = new QueueingBasicConsumer(_model);
	    var msgCount = GetMessageCount(_model, QueueName);

	    _model.BasicConsume(QueueName, true, consumer);

	    var count = 0;

	    while (count < msgCount)
	    {
	        var message = (Payment)consumer.Queue.Dequeue().Body.Deserialize(typeof(Payment));

	        Console.WriteLine("<=== Received {0} {1} {2}", message.CardNumber, message.AmountToPay, message.Name);
	        count++;
	    }
	}

	private static uint GetMessageCount(IModel channel, string queueName)
	{
	    var results = channel.QueueDeclare(queueName, true, false, false, null);

	    return results.MessageCount;
	}

	private static void SendMessage(Payment payment)
	{
	    _model.BasicPublish("", QueueName, null, payment.Serialize());
	    Console.WriteLine("===> Payment message sent: {0} {1} {2}", payment.CardNumber, payment.AmountToPay, payment.Name);
	}

	private static void CreateQueue()
	{
	    _connectionFactory = new ConnectionFactory() { HostName = "localhost", UserName = "test", Password = "test" };
	    _connection = _connectionFactory.CreateConnection();
	    _model = _connection.CreateModel();

	    _model.QueueDeclare(QueueName, true, false, false, null);
	}
}

In the code above, the `Main` method first mimics that 10 payments have been requested by customers which is following by `CreateQueue` method.

This method uses `ConnectionFactory` to setup a connect for future use by providing hostname, username and password details. For this demo, I’m using local RabbitMQ server so, I’m using `localhost` and `test` username that I created using RabbitMQ command prompt. This factory is then used to actually create a connection and a model. Model is then used to declare a queue named `StandardQueue` which makes sure that if such a queue already exists, it doesn’t create it again. The second parameter marks this queue as durable by specifying `true` which ensures that even if the server/machine is restarted, the same queue will be available as it will be written to the disk.

Next, in the main method, for every payment, message is sent using `SendMessage` method. Line no. 57 uses `BasicPublish` method to directly connect to the queue without specifying any exchange and send the payment object as `byte[]`. Behind the scenes, the default direct exchange is used to route to the queue.

If you put a breakpoint just at `Receive` method and navigate to RabbitMQ management console, you can see the 10 payments message sent to the queue.

When `Receive` method is called, the same application acts as a consumer. It specifies the queue name it would like to get messages from and it then dequeue messages one by one and deserialize them as well on line no. 43 to print the received message in the console.

Once the messages are dequeued, the messages from the queue is removed

Likewise, any number of messages can be sent from a publisher and a consumer can check the message broker for new messages at particular interval (fetch/pull) or it can subscribe to a queue. Hope this basic example explains how to work with standard queue in RabbitMQ. If you would like to learn more, read my other posts in Message Broker Software category and this course that has helped me to learn the concepts more in detail.