Introduction
In my previous blog, we saw how to create an Azure service bus queue on Azure Portal. Here in this blog, we will see how to send and receive messages from this queue using the .NET Standard library.
Getting Started
The service bus queues are azure messaging service that works as a brokered messaging communication model that is used for send and receive azure message bus from the azure service bus queue. Instead of communicating directly in the distributed application the components exchange messages via a queue, which acts as an intermediary (broker)
A message producer (sender) hands off a message to the queue and then continues its processing. Asynchronously, a message consumer (receiver) pulls the message from the queue and processes it. The producer does not have to wait for a reply from the consumer in order to continue to process and send further messages.
The Azure service bus queue offers First In, First Out (FIFO) message delivery to one or more competing consumers. That is, messages are typically received and processed by the receivers in the order in which they were added to the queue, and each message is received and processed by only one message consumer.
Demonstration
Before continue this make sure that you have Azure subscription, if you don’t have Azure subscription then don’t worry you can create a free account in the Azure portal. Here in the below demonstration to send and receive messages from the queue, we will write a C# console application with .net core using Visual Studio. Note that the following namespaces will be required in this demo project to make the codes runnable, hence add the reference of the namespaces during the demonstration.
- using System;
- using System.Text;
- using System.Threading;
- using System.Threading.Tasks;
- using Microsoft.Azure.ServiceBus;
Follow the below steps to send and receive messages from Azure service bus queue.
Launch Visual Studio and create a new Console App (.NET Core) project for C#.
Right-click the newly created project and select Manage NuGet Packages.
Select Browse. Search for and select Microsoft.Azure.ServiceBus.
Select Install to complete the installation, then close the NuGet Package Manager.
Use namespace Microsoft.Azure.ServiceBus; in the program.cs class.
Declare the following variables to store your connection string for the namespace, Quenue name and the create object of IQueueClient in your Programm class.
const string ServiceBusConnectionString = "<connection_string>"; const string QueueName = "<queue_name>"; static IQueueClient queueClient;Replace the Main() method with the following async Main method. It calls the send and receive methods of Queue that you will add in the next step to send and receive messages to and from the queue.
public static async Task Main(string[] args) { Console.WriteLine("======================================================"); Console.WriteLine("Press ENTER key to exit after sending and receiving the messages."); Console.WriteLine("======================================================"); const int numberOfMessages = 10; queueClient = new QueueClient(ServiceBusConnectionString, QueueName); // Send messages. await SendMessagesAsync(numberOfMessages); //Receive Messages //RegisterOnMessageHandlerAndReceiveMessages(); Console.ReadKey(); await queueClient.CloseAsync(); }Add the following SendMessagesAsync() method that does the work of sending the number of messages specified by numberOfMessagesToSend
static async Task SendMessagesAsync(int numberOfMessagesToSend) { try { for (var i = 0; i < numberOfMessagesToSend; i++) { // Create a new message to send to the queue. string messageBody = $"Message {i}"; var message = new Message(Encoding.UTF8.GetBytes(messageBody)); // Write the body of the message to the console. Console.WriteLine($"Sending message: {messageBody}"); // Send the message to the queue. await queueClient.SendAsync(message); } } catch (Exception exception) { Console.WriteLine($"{DateTime.Now} :: Exception: {exception.Message}"); } }Run the program, go to the Azure portal and check the message queue by selecting the name of your queue in the namespace Overview window to display queue Essentials.
If you are getting any error, refer to the source code given in the below of this blog.
Again add the following method, which registers the message handler and receives the messages sent by the sender application.
static void RegisterOnMessageHandlerAndReceiveMessages() { // Configure the message handler options in terms of exception handling, number of concurrent messages to deliver, etc. var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler) { // Maximum number of concurrent calls to the callback ProcessMessagesAsync(), set to 1 for simplicity. // Set it according to how many messages the application wants to process in parallel. MaxConcurrentCalls = 1, // Indicates whether the message pump should automatically complete the messages after returning from user callback. // False below indicates the complete operation is handled by the user callback as in ProcessMessagesAsync(). AutoComplete = false }; // Register the function that processes messages. queueClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions); }add the following ProcessMessagesAsync() method to process the received messages
static async Task ProcessMessagesAsync(Message message, CancellationToken token) { // Process the message. Console.WriteLine($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}"); // Complete the message so that it is not received again. // This can be done only if the queue Client is created in ReceiveMode.PeekLock mode (which is the default). await queueClient.CompleteAsync(message.SystemProperties.LockToken); // Note: Use the cancellationToken passed as necessary to determine if the queueClient has already been closed. // If queueClient has already been closed, you can choose to not call CompleteAsync() or AbandonAsync() etc. // to avoid unnecessary exceptions. }Add the ExceptionReceivedHandler() method to handle any exceptions that might occur.
static Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs) { Console.WriteLine($"Message handler encountered an exception {exceptionReceivedEventArgs.Exception}."); var context = exceptionReceivedEventArgs.ExceptionReceivedContext; Console.WriteLine("Exception context for troubleshooting:"); Console.WriteLine($"- Endpoint: {context.Endpoint}"); Console.WriteLine($"- Entity Path: {context.EntityPath}"); Console.WriteLine($"- Executing Action: {context.Action}"); return Task.CompletedTask; } }Finally, call the RegisterOnMessageHandlerAndReceiveMessages() method by uncommenting in the Main method to receive messages and run your application.
If you are getting any error, refer to the source code given in the below.
using Microsoft.Azure.ServiceBus;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace azurebusexplorer
{
class Program
{
const string ServiceBusConnectionString = "<your_connection_string>";
const string QueueName = "<your_queue_name>";
static IQueueClient queueClient;
public static async Task Main(string[] args)
{
Console.WriteLine("======================================================");
Console.WriteLine("Press ENTER key to exit after sending and receiving the messages.");
Console.WriteLine("======================================================");
const int numberOfMessages = 10;
queueClient = new QueueClient(ServiceBusConnectionString, QueueName);
// Send messages.
await SendMessagesAsync(numberOfMessages);
//Receive Messages
RegisterOnMessageHandlerAndReceiveMessages();
Console.ReadKey();
await queueClient.CloseAsync();
}
static async Task SendMessagesAsync(int numberOfMessagesToSend)
{
try
{
for (var i = 0; i < numberOfMessagesToSend; i++)
{
// Create a new message to send to the queue.
string messageBody = $"Message {i}";
var message = new Message(Encoding.UTF8.GetBytes(messageBody));
// Write the body of the message to the console.
Console.WriteLine($"Sending message: {messageBody}");
// Send the message to the queue.
await queueClient.SendAsync(message);
}
}
catch (Exception exception)
{
Console.WriteLine($"{DateTime.Now} :: Exception: {exception.Message}");
}
}
static void RegisterOnMessageHandlerAndReceiveMessages()
{
// Configure the message handler options in terms of exception handling, number of concurrent messages to deliver, etc.
var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
{
// Maximum number of concurrent calls to the callback ProcessMessagesAsync(), set to 1 for simplicity.
// Set it according to how many messages the application wants to process in parallel.
MaxConcurrentCalls = 1,
// Indicates whether the message pump should automatically complete the messages after returning from user callback.
// False below indicates the complete operation is handled by the user callback as in ProcessMessagesAsync().
AutoComplete = false
};
// Register the function that processes messages.
queueClient.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
}
static async Task ProcessMessagesAsync(Message message, CancellationToken token)
{
// Process the message.
Console.WriteLine($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");
// Complete the message so that it is not received again.
// This can be done only if the queue Client is created in ReceiveMode.PeekLock mode (which is the default).
await queueClient.CompleteAsync(message.SystemProperties.LockToken);
// Note: Use the cancellationToken passed as necessary to determine if the queueClient has already been closed.
// If queueClient has already been closed, you can choose to not call CompleteAsync() or AbandonAsync() etc.
// to avoid unnecessary exceptions.
}
// Use this handler to examine the exceptions received on the message pump.
static Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
Console.WriteLine($"Message handler encountered an exception {exceptionReceivedEventArgs.Exception}.");
var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
Console.WriteLine("Exception context for troubleshooting:");
Console.WriteLine($"- Endpoint: {context.Endpoint}");
Console.WriteLine($"- Entity Path: {context.EntityPath}");
Console.WriteLine($"- Executing Action: {context.Action}");
return Task.CompletedTask;
}
}
}
Summary
In the above of this blog, we saw how to receive and send message to Azure service bus queue, I hope you have enjoyed it a lot.
Thanks