​Working With Queue - Azure Service Bus

Introduction

 

Azure Service Bus is a fully managed multi-tenant cloud messaging service. It is an enterprise integration message broker. It is used to decouple the application and service from each other. The service bus queues support a brokered messaging model of communication. In this model, the components of a distributed application communicate with each other via Queues. Queue acts as a broker (intermediary). The message sender hands-off the message to the Queue and does not wait for the reply from the message receiver or consumer. Queue supports FIFO (First in, First out) message delivery to one or more receivers. Each message is received and processed by only one receiver. Queue helps us to scale our applications more easily. Once the consumer reads the message, it gets deleted from the Queue.

 

 

The namespace is a scoping container for all message components. A queue is created under the namespace. You can read my previous article to learn what Azure Service Bus is and how to create a namespace.

 

Create a Queue in the Azure portal

 

Namespace is the scoping container for message components, so Queue is created under the namespace. To create a queue, click on the "Queue" button. We can view all the queues under namespace in "Entities >> Queues" tab. On the "Create new queue" screen, there are multiple fields that need to be filled up, such as name, max queue size, message time to live, lock duration, and other option.

 

 

In the following example, I have created one sender and one receiver. The sender sends the message to the queue and the receiver reads a message from the queue and processes it. Here, I am using .NET Core console application to send and receive the message.

 

At the initial part of the code, I have created an instance of QueueClient using a connnection string and a queue name. The QueueClient is available in Microsoft.Azure.ServiceBus NuGet package. It can be downloaded using the NuGet Package Manager.

 

  

Using the SendAsync method of queue client, we can send the message to the queue.

 

Sender Code

static IQueueClient queueClient;
static void Main(string[] args)
{
    string connectionStringServiceBus = "<<Namespace connection string>>";
    string queueName = "<<queue name>>";

    SendMessage(connectionStringServiceBus, queueName).GetAwaiter().GetResult();
}

static async Task SendMessage(string connectionStringServiceBus, string queueName)
{
    const int numberOfMessages = 5;
    queueClient = new QueueClient(connectionStringServiceBus, queueName);

    Console.WriteLine("======================================================");
    Console.WriteLine("Press any key to exit after sending all the messages.");
    Console.WriteLine("======================================================");

    // Send Messages
    await SendMessagesToQueueAsync(numberOfMessages);

    Console.ReadKey();

    await queueClient.CloseAsync();
}

static async Task SendMessagesToQueueAsync(int numberOfMessages)
{
    try
    {
        for (var i = 0; i < numberOfMessages; i++)
        {
            // Message that send to the queue
            string messageBody = $"GNR-MSG dataid:{i}";
            var message = new Message(Encoding.UTF8.GetBytes(messageBody));

            Console.WriteLine($"Sending message to queue: {messageBody}");

            // Send the message to the queue
            await queueClient.SendAsync(message);
        }
    }
    catch (Exception exception)
    {
        Console.WriteLine($"Exception: {exception.Message}");
    }
}

The RegisterMessageHandler method of queue is used to register the function that will process the message. This method expects two parameters: handler and message handler options. The handler processes the message.

 

Receiver Code

static IQueueClient queueClient;
static void Main(string[] args)
{
    string connectionStringServiceBus = "<<Namespace connection string>>";
    string queueName = "<<queue name>>";

    Console.WriteLine("======================================================");
    Console.WriteLine("Press any key to exit after receiving all the messages.");
    Console.WriteLine("======================================================");

    queueClient = new QueueClient(connectionStringServiceBus, queueName);

    RegisterMessageHandlerAndReceiveMessages();

    Console.ReadKey();

    queueClient.CloseAsync().Wait();
}
static void RegisterMessageHandlerAndReceiveMessages()
{
    // Configure the MessageHandler Options in terms of exception handling, number of concurrent messages to deliver etc.
    var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
    {
        MaxConcurrentCalls = 1,
        AutoComplete = false
    };

    // Register the function that will process messages
    queueClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
}

static async Task ProcessMessagesAsync(Message message, CancellationToken token)
{
    // Process the message
    Console.WriteLine($"Received message: Sequence Number:{message.SystemProperties.SequenceNumber} \t Body:{Encoding.UTF8.GetString(message.Body)}");

    await queueClient.CompleteAsync(message.SystemProperties.LockToken);
}
// Use this Handler to look at the exceptions received on the MessagePump
static Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
    Console.WriteLine($"Exception:: {exceptionReceivedEventArgs.Exception}.");
    return Task.CompletedTask;
}

Output

 

 

When we run the sender application and check on the Azure portal under Queue >> Overview tab, the following screen is displayed. Here, we can see that "Active Message Count" has a value of 5. This value denotes the number of messages in the queue. It also contains the current size of the messages.

 

 

Summary

 

A Queue is used to send and receive messages. It enables us to store these messages until the receiver is available to receive such messages and process them. It allows one-way communication. The messages are in FIFO (first in, first out) order in the queue and are timestamped based on their arrival. The queue is used for point-to-point communication.

 

You can view and download the source code from GitHub.