Azure Event Hubs Documentation

Step 1: Create an Azure Event Hubs Namespace

An Event Hubs namespace is a logical container for Event Hubs resources. You need a namespace before you can create an Event Hub.

1. Sign in to the Azure portal.
2. In the Azure portal, search for and select Event Hubs.
3. On the Event Hubs page, select Create.
4. On the Create namespace page, select or fill in the following fields:
   • Subscription: Select your Azure subscription.
   • Resource group: Select or create a resource group.
   • Namespace name: Enter a unique name for your namespace.
   • Location: Select the Azure region for your namespace.
   • Pricing tier: Choose a pricing tier (e.g., Basic, Standard). For getting started, Standard is often a good choice.
5. Select Review + create, then Create.

It may take a few minutes for the namespace to be deployed.

Step 2: Create an Event Hub

Once your namespace is deployed, you can create an Event Hub within it. An Event Hub is the actual entity that holds the stream of events.

1. Navigate to your newly created Event Hubs namespace in the Azure portal.
2. In the left-hand menu, under Entities, select Event hubs.
3. Click + Event Hub.
4. Enter a name for your Event Hub (e.g., myeventhub).
5. Configure settings like Partition count and Message retention. For basic testing, default values are often sufficient.
6. Click Create.

Step 3: Get Connection Strings

To connect to your Event Hub, you'll need a connection string. These strings contain the authorization credentials and endpoint information.

1. In your Event Hubs namespace, navigate to Shared access policies under Access.
2. Select the RootManageSharedAccessKey policy (or create a new one with appropriate permissions).
3. Copy the Primary Connection String. This connection string will be used by your applications to send and receive data.

Step 4: Send and Receive Events

Now you can start sending and receiving events. We'll demonstrate with a simple C# example using the Azure SDK for .NET. You can adapt this to other languages like Python, Java, or Node.js.

Prerequisites:

• .NET SDK installed.
• A C# project (e.g., a Console Application).

Install the Azure Event Hubs SDK:

dotnet add package Azure.Messaging.EventHubs
dotnet add package Azure.Messaging.EventHubs.Processor
                

Example: Sending Events (Producer)

Replace YOUR_EVENTHUB_CONNECTION_STRING and myeventhub with your actual values.

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

public class EventHubProducer
{
    // Replace with your Event Hubs connection string and hub name
    static string connectionString = "YOUR_EVENTHUB_CONNECTION_STRING";
    static string eventHubName = "myeventhub";

    public static async Task SendEventsAsync(int numberOfEvents)
    {
        await using var producerClient = new EventHubProducerClient(connectionString, eventHubName);

        Console.WriteLine($"Sending {numberOfEvents} events...");

        for (int i = 0; i < numberOfEvents; i++)
        {
            var eventData = new EventData(Encoding.UTF8.GetBytes($"Event {i + 1}"));
            try
            {
                await producerClient.SendAsync(eventData);
                Console.WriteLine($"Sent: Event {i + 1}");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Error sending event {i + 1}: {ex.Message}");
            }
        }
        Console.WriteLine("Finished sending events.");
    }

    public static async Task Main(string[] args)
    {
        await SendEventsAsync(10); // Send 10 events
    }
}
                

Example: Receiving Events (Consumer)

This example uses the Event Processor Client for a more robust consumer experience, managing partitions and checkpoints.

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

public class EventHubConsumer
{
    // Replace with your Event Hubs connection string and hub name
    static string eventHubConnectionString = "YOUR_EVENTHUB_CONNECTION_STRING";
    static string eventHubName = "myeventhub";
    // Replace with your Blob Storage connection string and container name
    static string blobStorageConnectionString = "YOUR_BLOB_STORAGE_CONNECTION_STRING";
    static string blobContainerName = "eventhub-checkpoints";

    public static async Task RunConsumerAsync()
    {
        string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
        string fullyQualifiedNamespace = "YOUR_EVENTHUBS_NAMESPACE.servicebus.windows.net"; // e.g., mynamespace.servicebus.windows.net

        var blobClientOptions = new BlobClientOptions();
        blobClientOptions.Retry.MaxRetries = 5;
        blobClientOptions.Retry.Delay = TimeSpan.FromSeconds(10);

        var blobServiceClient = new BlobServiceClient(blobStorageConnectionString, blobClientOptions);
        var blobContainerClient = blobServiceClient.GetBlobContainerClient(blobContainerName);

        var eventProcessorOptions = new EventProcessorClientOptions
        {
            MaximumReceiveBatchSize = 100,
            PrefetchCount = 200
        };

        var processor = new EventProcessorClient(blobContainerClient, consumerGroup, eventHubConnectionString, eventHubName, eventProcessorOptions);

        processor.ProcessEventAsync += (args) =>
        {
            Console.WriteLine($"Received event: Sequence Number={args.Data.SequenceNumber}, PartitionId={args.PartitionId}");
            Console.WriteLine($"\tBody: {Encoding.UTF8.GetString(args.Data.EventBody.ToArray())}");
            return Task.CompletedTask;
        };

        processor.ProcessErrorAsync += (args) =>
        {
            Console.WriteLine($"\tERROR: {args.Exception.Message}");
            return Task.CompletedTask;
        };

        Console.WriteLine("Starting Event Processor...");
        await processor.StartProcessingAsync();

        Console.WriteLine("Press Enter to stop the processor.");
        Console.ReadLine();

        Console.WriteLine("Stopping Event Processor...");
        await processor.StopProcessingAsync();
        Console.WriteLine("Event Processor stopped.");
    }

    public static async Task Main(string[] args)
    {
        // Ensure you have created the Blob Storage container beforehand.
        // You also need to replace placeholders with your actual connection strings and namespace.
        await RunConsumerAsync();
    }
}