クイックスタート: Go を使用して Event Hubs との間でイベントを送受信する

[アーティクル]
06/26/2023

Azure Event Hubs はビッグデータストリーミングプラットフォームであり、毎秒数百万のイベントを受け取って処理できるイベントインジェストサービスです。 Event Hubs では、分散されたソフトウェアやデバイスから生成されるイベント、データ、またはテレメトリを処理および格納できます。イベントハブに送信されたデータは、任意のリアルタイム分析プロバイダーやバッチ処理/ストレージアダプターを使用して、変換および保存できます。 Event Hubs の詳しい概要については、Event Hubs の概要と Event Hubs の機能に関するページをご覧ください。

このクイックスタートでは、イベントハブとの間でイベントを送受信する Go アプリケーションを作成する方法について説明します。

注意

このクイックスタートは、https://github.com/Azure/azure-sdk-for-go/tree/main/sdk/messaging/azeventhubs にある GitHub 上のサンプルに基づいています。「イベントの送信」セクションは example_producing_events_test.go サンプルに基づいており、受信は example_processor_test.go サンプルに基づいています。コードはクイックスタート用に簡略化されており、すべての詳細コメントは削除されているため、詳細と説明についてはサンプルを参照してください。

前提条件

このクイックスタートを完了するには、次の前提条件を用意しておく必要があります。

Go がローカルにインストールされていること。必要に応じて、こちらの手順に従います。
アクティブな Azure アカウントアカウントがない場合、Azure 試用版にサインアップして、最大 10 件の無料 Mobile Apps を入手できます。 Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。
Event Hubs 名前空間とイベントハブを作成する。 Azure portal を使用して Event Hubs 型の名前空間を作成し、アプリケーションがイベントハブと通信するために必要な管理資格情報を取得します。名前空間とイベントハブを作成するには、こちらの記事の手順に従います。

送信イベント

このセクションでは、イベントハブにイベントを送信する Go アプリケーションの作成方法を説明します。

Go パッケージをインストールする

次の例に示すように、Event Hubs 用の Go パッケージを取得します。

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

イベントハブにイベントを送信するためのコード

イベントハブにイベントを送信するためのコードを次に示します。このコードでの主な手順は次のとおりです。

Event Hubs 名前空間への接続文字列とイベントハブ名を使用して、Event Hubs のプロデューサークライアントを作成します。
バッチオブジェクトを作成し、バッチにサンプルイベントを追加します。
イベントのバッチをイベントハブに送信します。

重要

NAMESPACE CONNECTION STRING を Event Hubs 名前空間への接続文字列に置き換え、EVENT HUB NAME をサンプルコード内のイベントハブ名に置き換えます。

package main

import (
	"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 hub
	producerClient, err := azeventhubs.NewProducerClientFromConnectionString("NAMESPACE CONNECTION STRING", "EVENT HUB NAME", nil)

	if err != nil {
		panic(err)
	}

	defer producerClient.Close(context.TODO())

	// create sample events
	events := createEventsForSample()

	// create a batch object and add sample events to the batch
	newBatchOptions := &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 hub
	err = producerClient.SendEventDataBatch(context.TODO(), batch, nil)

	if err != nil {
		panic(err)
	}
}

func createEventsForSample() []*azeventhubs.EventData {
	return []*azeventhubs.EventData{
		{
			Body: []byte("hello"),
		},
		{
			Body: []byte("world"),
		},
	}
}

アプリケーションはまだ実行しないでください。最初に受信側アプリを実行し、次に送信者アプリを実行する必要があります。

受信イベント

Storage アカウントとコンテナーの作成

パーティションのリースやイベント内のチェックポイントなどの状態は、Azure Storage コンテナーを使用して受信者間で共有されます。 Go SDK を使用してストレージアカウントとコンテナーを作成できますが、「Azure ストレージアカウントについて」の手順に従って作成することもできます。

チェックポイントストアとして Azure Blob Storage を使用する場合は、次の推奨事項に従ってください。

コンシューマーグループごとに個別のコンテナーを使用します。同じストレージアカウントを使用できますが、各グループごとに 1 つのコンテナーを使用します。
コンテナーを他の何かに使用しないでください。また、ストレージアカウントも他の何かに使用しないでください。
ストレージアカウントは、デプロイされたアプリケーションが配置されているのと同じリージョンに存在する必要があります。アプリケーションがオンプレミスの場合は、可能な中で最も近いリージョンを選択することを試みてください。

Azure portal の [ストレージアカウント] ページの [Blob service] セクションで、次の設定が無効になっていることを確認してください。

階層型名前空間
BLOB の論理的な削除
バージョン管理

Go パッケージ

メッセージを受信するには、次の例に示すように、Event Hubs 用の Go パッケージを取得します。

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

イベントハブからイベントを受信するためのコード

イベントハブからイベントを受信するためのコードを次に示します。このコードでの主な手順は次のとおりです。

イベントハブでチェックポイント処理に使用される Azure Blob Storage を表すチェックポイントストアオブジェクトを確認します。
Event Hubs 名前空間への接続文字列とイベントハブ名を使用して、Event Hubs のコンシューマークライアントを作成します。
クライアントオブジェクトとチェックポイントストアオブジェクトを使用してイベントプロセッサを作成します。プロセッサによってイベントが受信されて処理されます。
イベントハブ内のパーティションごとに、イベントを処理する関数として processEvents を使用してパーティションクライアントを作成します。
すべてのパーティションクライアントを実行して、イベントを受信して処理します。

重要

次のプレースホルダーの値を実際の値に置き換えます。

AZURE STORAGE CONNECTION STRING を、Azure ストレージアカウントの接続文字列に
BLOB CONTAINER NAME を、ストレージアカウントで作成した BLOB コンテナーの名前に
NAMESPACE CONNECTION STRING を、Event Hubs 名前空間への接続文字列に
EVENT HUB NAME を、サンプルコードのイベントハブ名に

package main

import (
	"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 name
	checkClient, 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 hub
	checkpointStore, err := checkpoints.NewBlobStore(checkClient, nil)

	if err != nil {
		panic(err)
	}

	// create a consumer client using a connection string to the namespace and the event hub
	consumerClient, 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 events
	processor, 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 events
	dispatchPartitionClients := func() {
		for {
			partitionClient := processor.NextPartitionClient(context.TODO())

			if partitionClient == nil {
				break
			}

			go func() {
				if err := processEvents(partitionClient); err != nil {
					panic(err)
				}
			}()
		}
	}

	// run all partition clients
	go 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())
}

受信側と送信側アプリを実行する

まず、受信側アプリを実行します。
送信側アプリを実行します。
受信側ウィンドウに次の出力が表示されるまで 1 分間待ちます。
```
Processing 2 event(s)
Event received with body hello
Event received with body world
```

次の手順

https://github.com/Azure/azure-sdk-for-go/tree/main/sdk/messaging/azeventhubs にある GitHub 上のサンプルを参照してください。

次の方法で共有

クイックスタート: Go を使用して Event Hubs との間でイベントを送受信する

前提条件

送信イベント

Go パッケージをインストールする

イベントハブにイベントを送信するためのコード

受信イベント

Storage アカウントとコンテナーの作成

Go パッケージ

イベントハブからイベントを受信するためのコード

受信側と送信側アプリを実行する

次の手順

フィードバック

その他のリソース

次の方法で共有

クイックスタート: Go を使用して Event Hubs との間でイベントを送受信する

前提条件

送信イベント

Go パッケージをインストールする

イベント ハブにイベントを送信するためのコード

受信イベント

Storage アカウントとコンテナーの作成

Go パッケージ

イベント ハブからイベントを受信するためのコード

受信側と送信側アプリを実行する

次の手順

フィードバック

その他のリソース

イベントハブにイベントを送信するためのコード

イベントハブからイベントを受信するためのコード