Sending and Receiving Messages with Azure Event Hubs

A practical guide to leveraging Azure Event Hubs for real-time data streaming.

Azure Event Hubs is a highly scalable data streaming platform and event ingestion service. It can capture millions of events per second from multiple sources, so you can build a variety of real-time analytics solutions.

This tutorial will walk you through the fundamental steps of sending messages to an Event Hub and then receiving them using a consumer application. We'll use Azure CLI and a simple C# example for demonstration.

Prerequisites

Step 1: Create an Event Hubs Namespace and Event Hub

First, we need a place to send our events. This is called an Event Hubs Namespace, which is a collection of Event Hubs. Then, within that namespace, we create the Event Hub itself.

Open your terminal or command prompt and run the following Azure CLI commands:

# Set your desired resource group and location
RESOURCE_GROUP="myEventHubResourceGroup"
LOCATION="eastus"
NAMESPACE="myUniqueEventHubsNamespace-$(date +%s)" # Ensure uniqueness
EVENTHUB_NAME="myEventHub"

# Create a resource group if it doesn't exist
az group create --name $RESOURCE_GROUP --location $LOCATION

# Create an Event Hubs namespace
az eventhubs namespace create --resource-group $RESOURCE_GROUP --name $NAMESPACE --location $LOCATION

# Create an Event Hub within the namespace
az eventhubs eventhub create --resource-group $RESOURCE_GROUP --namespace-name $NAMESPACE --name $EVENTHUB_NAME

Note down the created namespace name, as you'll need it later.

Step 2: Get Connection String

Your applications will need a connection string to authenticate with Event Hubs. You can retrieve this using the Azure CLI.

az eventhubs authorization rule list --resource-group $RESOURCE_GROUP --namespace-name $NAMESPACE --query "[?contains(name, 'RootManageSharedAccessKey')].{keyName:name, primaryKey:primaryKey}" -o json

# You can also get the full connection string directly for a specific hub
az eventhubs eventhub show --resource-group $RESOURCE_GROUP --namespace-name $NAMESPACE --name $EVENTHUB_NAME --query "authorizationRules[?contains(name, 'RootManageSharedAccessKey')].{connectionString:primaryConnectionString}" -o tsv

Copy the primaryConnectionString. This will be used in our sample applications.

Step 3: Sending Messages (C# Example)

Let's create a simple C# console application to send messages to our Event Hub.

Project Setup

Create a new .NET Core console application:

dotnet new console -n EventHubSender
cd EventHubSender

Add the Azure Event Hubs NuGet package:

dotnet add package Azure.Messaging.EventHubs

Sender Code

Replace the contents of Program.cs with the following code. Remember to replace YOUR_EVENTHUB_CONNECTION_STRING with the connection string you obtained in Step 2.

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

class Program
{
    // Replace with your actual Event Hub connection string
    private const string connectionString = "YOUR_EVENTHUB_CONNECTION_STRING";
    private const string eventHubName = "myEventHub"; // Or the name of your Event Hub

    static async Task Main(string[] args)
    {
        Console.WriteLine("Sending messages to Event Hub...");

        // Create an EventHubProducerClient
        await using var producer = new EventHubProducerClient(connectionString, eventHubName);

        // Create a batch of events
        using EventDataBatch eventDataBatch = await producer.CreateBatchAsync();

        for (int i = 1; i <= 5; i++)
        {
            var message = $"Message {i} from EventHubSender";
            var eventData = new EventData(Encoding.UTF8.GetBytes(message));

            if (!eventDataBatch.TryAdd(eventData))
            {
                // If the batch is full, send it and create a new one
                await producer.SendAsync(eventDataBatch);
                Console.WriteLine($"Sent a batch of {eventDataBatch.Count} events.");
                // Recreate the batch
                eventDataBatch = await producer.CreateBatchAsync();
                // Try adding the event again to the new batch
                if (!eventDataBatch.TryAdd(eventData))
                {
                    Console.WriteLine($"Event {i} was too large and cannot be added to the batch.");
                }
            }
        }

        // Send the last batch if it contains any events
        if (eventDataBatch.Count > 0)
        {
            await producer.SendAsync(eventDataBatch);
            Console.WriteLine($"Sent the final batch of {eventDataBatch.Count} events.");
        }

        Console.WriteLine("Messages sent successfully!");
    }
}

Run the sender application:

dotnet run

Step 4: Receiving Messages (C# Example)

Now, let's create a C# console application to receive the messages we just sent.

Project Setup

Create another .NET Core console application:

dotnet new console -n EventHubReceiver
cd EventHubReceiver

Add the Azure Event Hubs NuGet package:

dotnet add package Azure.Messaging.EventHubs

Receiver Code

Replace the contents of Program.cs with the following code. Again, replace YOUR_EVENTHUB_CONNECTION_STRING with your actual connection string.

C#
using System;
using System.Text;
using System.Threading.Tasks;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;

class Program
{
    // Replace with your actual Event Hub connection string
    private const string connectionString = "YOUR_EVENTHUB_CONNECTION_STRING";
    private const string eventHubName = "myEventHub"; // Or the name of your Event Hub
    private const string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName; // Using the default consumer group

    static async Task Main(string[] args)
    {
        Console.WriteLine("Starting to receive messages from Event Hub...");

        // Create an EventHubConsumerClient
        await using var consumer = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName);

        Console.CancelKeyPress += (sender, eventArgs) =>
        {
            eventArgs.Cancel = true; // Prevent the process from terminating immediately
            Console.WriteLine("Cancellation requested. Stopping receiver...");
            // Add logic here to gracefully shut down the consumer if needed
        };

        try
        {
            // Begin reading events from all partitions
            await foreach (PartitionEvent partitionEvent in consumer.ReadEventsAsync())
            {
                string messageBody = Encoding.UTF8.GetString(partitionEvent.Data.EventBody.ToArray());
                Console.WriteLine($"Received message: '{messageBody}' from partition {partitionEvent.Partition.Id}");
            }
        }
        catch (TaskCanceledException)
        {
            Console.WriteLine("Receiver task was canceled.");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"An error occurred: {ex.Message}");
        }
        finally
        {
            Console.WriteLine("Event Hub receiver stopped.");
        }
    }
}

Run the receiver application before or after the sender. If you run it after, it might take a moment for events to appear if it missed the initial send. If you run it before, it will wait for messages.

dotnet run

You should see the messages you sent appear in the receiver's console output.

Next Steps