Azure Event Hubs with .NET

Event-driven architectures are fundamental in modern application development, especially when handling massive streams of data. Azure Event Hubs serves as a highly scalable data streaming platform and event ingestion service, which can process millions of events per second.

In this article, we’ll explore how to effectively use Azure Event Hubs with .NET, guiding you through the setup, sending events, and receiving them in a clean and efficient manner.

Azure — Event Hubs v/s Queues v/s Service Bus

Understanding the key characteristics for these Azure services before diving into this article will help us to understand better.

Azure Event Hubs

Purpose: Azure Event Hubs is a big data streaming platform and event ingestion service, capable of receiving and processing millions of events per second. Event Hubs can process and store events, data, or telemetry produced by distributed software and devices.

Use Case: Event Hubs is ideal for event-driven architectures, telemetry, data streaming, and processing pipelines where high throughput and real-time data processing are needed. For example, it’s well-suited for logging data from multiple sources, aggregating data like sensor readings, or handling streaming from mobile apps.

Key Characteristics:

• Handles massive amounts of events or messages.
• Supports multiple concurrent readers.
• Provides low-latency, capable of processing and making events available within milliseconds.
• Retains data for a configured period, defaulting to a retention of 1 to 7 days, which can be useful for replaying events if needed.

Azure Queues (Queue Storage)

Purpose: Azure Queue Storage is part of Azure’s storage infrastructure, providing reliable messaging between application components. Queues are used for storing large numbers of messages that can be accessed from anywhere in the world via authenticated calls using HTTP or HTTPS.

Use Case: Azure Queues are best for making smooth communication between parts of a cloud service, or between different services running in Azure. This can be useful for workload balancing and to manage bursts of high load, like scaling out web application requests to background workers.

Key Characteristics:

• Simple and easy to use.
• Designed to handle high volumes of messages, though with lower throughput compared to Event Hubs.
• Each message is processed by a single consumer.
• Provides a way to decouple components in a system.

Azure Service Bus Topics and Subscriptions

Purpose: Topics and subscriptions are a part of Azure Service Bus, which is an enterprise messaging system as a service. It supports more complex messaging features like FIFO (First-In, First-Out) messaging, ordered delivery, deduplication, and sessions.

Use Case: Service Bus Topics are ideal for publish/subscribe scenarios where each message can be subscribed to by multiple consumers. It is used extensively in enterprise-level applications where communication between decoupled systems is required across different components of the business.

Key Characteristics:

• Supports multiple, independent subscribers to the same message.
• Each subscription can optionally filter messages.
• More features for handling sophisticated messaging patterns and transactional processing.
• Ensures advanced security, reliability, and scaling capabilities that are above those provided by Azure Queue Storage.

Setting Up Your Environment

You’ll need an active Microsoft Azure subscription and Visual Studio 2022, as the Azure Event Hubs client library utilizes features from C# 8.0 and above.

The first step is to create an Event Hubs namespace and an event hub within the Azure portal. This namespace acts as a container for all your event hubs.

Coding the Event Producer

Once your environment is set up, begin by creating a .NET Core console application in Visual Studio 2022.

This application will be responsible for sending events to the Event Hub. Here’s a simplified version of the code you might write, focusing on clarity and functionality:

using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
using System;
using System.Text;
using System.Threading.Tasks;

public class Program
{
    private const string EventHubNamespace = "your_event_hub_namespace.servicebus.windows.net";
    private const string HubName = "your_event_hub_name";
    
    public static async Task Main()
    {
        var producerClient = new EventHubProducerClient(EventHubNamespace, HubName, new DefaultAzureCredential());

        using var eventBatch = await producerClient.CreateBatchAsync();
        for (int i = 1; i <= 3; i++)
        {
            if (!eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes($"Event {i}"))))
            {
                Console.WriteLine($"Event {i} is too large for the batch and cannot be sent.");
            }
        }

        try
        {
            await producerClient.SendAsync(eventBatch);
            Console.WriteLine("A batch of 3 events has been published.");
        }
        finally
        {
            await producerClient.DisposeAsync();
        }
    }
}

This code snippet demonstrates:

• EventHubProducerClient is initialized using the namespace URL of the Event Hub and the specific Event Hub name. Authentication is managed by DefaultAzureCredential, which simplifies securing application secrets.
• CreateBatchAsync method is called to start a batch of events. This helps in optimizing network and resource utilization by grouping multiple events into a single send operation.
• Inside a loop, individual events are created from strings ("Event {i}"), converted to bytes, and then added to the batch. The TryAdd method returns false if the event cannot be added because the batch is full or exceeds the size limit.
• SendAsync sends the entire batch of events to the Event Hub. Error handling ensures that the client is properly disposed of after sending, whether successful or in case of an exception.

Building the Event Consumer

To receive events, you’ll also create a .NET Core console application. This application will use the EventProcessorClient, which simplifies the processing of events coming from Azure Event Hubs:

using Azure.Messaging.EventHubs.Consumer;
using Azure.Messaging.EventHubs.Processor;
using Azure.Storage.Blobs;
using System;
using System.Text;
using System.Threading.Tasks;

public class Program
{
    private const string BlobStorageConnectionString = "your_blob_storage_connection_string";
    private const string BlobContainerName = "your_blob_container_name";
    private const string EventHubConnectionString = "your_event_hub_connection_string";
    private const string EventHubName = "your_event_hub_name";
    
    public static async Task Main()
    {
        var storageClient = new BlobContainerClient(BlobStorageConnectionString, BlobContainerName);
        var processor = new EventProcessorClient(storageClient, EventHubConsumerClient.DefaultConsumerGroupName, EventHubConnectionString, EventHubName);

        processor.ProcessEventAsync += ProcessEventHandler;
        processor.ProcessErrorAsync += ProcessErrorHandler;

        await processor.StartProcessingAsync();

        Console.WriteLine("Waiting for events...");
        
        await Task.Delay(TimeSpan.FromMinutes(1));
        await processor.StopProcessingAsync();
    }

    static Task ProcessEventHandler(ProcessEventArgs eventArgs)
    {
        Console.WriteLine($"Received event: {Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray())}");
        return Task.CompletedTask;
    }

    static Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
    {
        Console.WriteLine($"Error on partition {eventArgs.PartitionId}: {eventArgs.Exception.Message}");
        return Task.CompletedTask;
    }
}

This code sets up:

• Initialized with a connection to Azure Blob Storage (used for checkpointing) and Event Hubs. Checkpointing helps to manage event reading state, so in case of a disruption, the consumer can continue from the last event it processed.
• ProcessEventAsync and ProcessErrorAsync are event handlers for successfully processed events and for processing errors, respectively.
• StartProcessingAsync begins the event processing. It runs continuously, waiting for events to arrive. The consumer handles each event as specified in the ProcessEventAsync handler.
• StopProcessingAsync is called to cleanly stop the event processor after a specified time.

This setup is designed to be both robust and scalable, handling errors gracefully and ensuring that event processing can recover from interruptions. Each component of the code plays a critical role in ensuring data flows efficiently and securely between your application components and Azure Event Hubs.

I trust this information has been valuable to you. 🌟 Wishing you an enjoyable and enriching learning journey!

📚 For more insights like these, feel free to 👏 follow 👉 Merwan Chinta