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. Data sent to an event hub can be transformed and stored using any real-time analytics provider or batching/storage adapters. For detailed overview of Event Hubs, seeEvent Hubs overview andEvent Hubs features.

This quickstart describes how to write Go applications to send events to or receive events from an event hub.

Prerequisites

To complete this quickstart, you need the following prerequisites:

• Go installed locally. Followthese instructions if necessary.
• An active Azure account. If you don't have an Azure subscription, create afree account before you begin.
• Create an Event Hubs namespace and an event hub. Use theAzure portal to create a namespace of type Event Hubs, and obtain the management credentials your application needs to communicate with the event hub. To create a namespace and an event hub, follow the procedure inthis article.

Send events

This section shows you how to create a Go application to send events to an event hub.

Install Go package

Get the Go package for Event Hubs as shown in the following example.

go get github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs

Code to send events to an event hub

Here's the code to send events to an event hub. The main steps in the code are:

1. Create an Event Hubs producer client using a connection string to the Event Hubs namespace and the event hub name.
2. Create a batch object and add sample events to the batch.
3. Send the batch of events to th events.

package mainimport ("context""github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs")func main() {// create an Event Hubs producer client using a connection string to the namespace and the event hubproducerClient, err := azeventhubs.NewProducerClientFromConnectionString("NAMESPACE CONNECTION STRING", "EVENT HUB NAME", nil)if err != nil {panic(err)}defer producerClient.Close(context.TODO())// create sample eventsevents := createEventsForSample()// create a batch object and add sample events to the batchnewBatchOptions := &azeventhubs.EventDataBatchOptions{}batch, err := producerClient.NewEventDataBatch(context.TODO(), newBatchOptions)if err != nil {panic(err)}for i := 0; i < len(events); i++ {err = batch.AddEventData(events[i], nil)if err != nil {panic(err)}}// send the batch of events to the event huberr = producerClient.SendEventDataBatch(context.TODO(), batch, nil)if err != nil {panic(err)}}func createEventsForSample() []*azeventhubs.EventData {return []*azeventhubs.EventData{{Body: []byte("hello"),},{Body: []byte("world"),},}}

Don't run the application yet. You first need to run the receiver app and then the sender app.

Receive events

Create a Storage account and container

State such as leases on partitions and checkpoints in the events are shared between receivers using an Azure Storage container. You can create a storage account and container with the Go SDK, but you can also create one by following the instructions inAbout Azure storage accounts.

Follow these recommendations when you use Azure Blob Storage as a checkpoint store:

• Use a separate container for each consumer group. You can use the same storage account, but use one container per each group.
• Don't use the storage account for anything else.
• Don't use the container for anything else.
• Create the storage account in the same region as the deployed application. If the application is on-premises, try to choose the closest region possible.

On theStorage account page in the Azure portal, in theBlob service section, ensure that the following settings are disabled.

• Hierarchical namespace
• Blob soft delete
• Versioning

Go packages

To receive the messages, get the Go packages for Event Hubs as shown in the following example.

go get github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubsgo get github.com/Azure/azure-sdk-for-go/sdk/storage/azblob

Code to receive events from an event hub

Here's the code to receive events from an event hub. The main steps in the code are:

1. Check a checkpoint store object that represents the Azure Blob Storage used by the event hub for checkpointing.
2. Create an Event Hubs consumer client using a connection string to the Event Hubs namespace and the event hub name.
3. Create an event processor using the client object and the checkpoint store object. The processor receives and processes events.
4. For each partition in the event hub, create a partition client with processEvents as the function to process events.
5. Run all partition clients to receive and process events.

package mainimport ("context""errors""fmt""time""github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs""github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/checkpoints""github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container")func main() {// create a container client using a connection string and container namecheckClient, err := container.NewClientFromConnectionString("AZURE STORAGE CONNECTION STRING", "CONTAINER NAME", nil)if err != nil {panic(err)}// create a checkpoint store that will be used by the event hubcheckpointStore, err := checkpoints.NewBlobStore(checkClient, nil)if err != nil {panic(err)}// create a consumer client using a connection string to the namespace and the event hubconsumerClient, err := azeventhubs.NewConsumerClientFromConnectionString("NAMESPACE CONNECTION STRING", "EVENT HUB NAME", azeventhubs.DefaultConsumerGroup, nil)if err != nil {panic(err)}defer consumerClient.Close(context.TODO())// create a processor to receive and process eventsprocessor, err := azeventhubs.NewProcessor(consumerClient, checkpointStore, nil)if err != nil {panic(err)}//  for each partition in the event hub, create a partition client with processEvents as the function to process eventsdispatchPartitionClients := func() {for {partitionClient := processor.NextPartitionClient(context.TODO())if partitionClient == nil {break}go func() {if err := processEvents(partitionClient); err != nil {panic(err)}}()}}// run all partition clientsgo dispatchPartitionClients()processorCtx, processorCancel := context.WithCancel(context.TODO())defer processorCancel()if err := processor.Run(processorCtx); err != nil {panic(err)}}func processEvents(partitionClient *azeventhubs.ProcessorPartitionClient) error {defer closePartitionResources(partitionClient)for {receiveCtx, receiveCtxCancel := context.WithTimeout(context.TODO(), time.Minute)events, err := partitionClient.ReceiveEvents(receiveCtx, 100, nil)receiveCtxCancel()if err != nil && !errors.Is(err, context.DeadlineExceeded) {return err}fmt.Printf("Processing %d event(s)\n", len(events))for _, event := range events {fmt.Printf("Event received with body %v\n", string(event.Body))}if len(events) != 0 {if err := partitionClient.UpdateCheckpoint(context.TODO(), events[len(events)-1], nil); err != nil {return err}}}}func closePartitionResources(partitionClient *azeventhubs.ProcessorPartitionClient) {defer partitionClient.Close(context.TODO())}

Run receiver and sender apps

1. Run the receiver app first.

2. Run the sender app.

3. Wait for a minute to see the following output in the receiver window.

   Processing 2 event(s)Event received with body helloEvent received with body world

Next steps

See samples on GitHub athttps://github.com/Azure/azure-sdk-for-go/tree/main/sdk/messaging/azeventhubs.