Mulai cepat: Mengirim peristiwa ke atau menerima peristiwa dari Azure Event Hubs menggunakan Go

Azure Event Hubs adalah platform streaming Big Data dan layanan pengolahan peristiwa, yang mampu menerima dan memproses jutaan peristiwa per detik. Azure Event Hubs dapat memproses dan menyimpan peristiwa, data, atau telemetri yang diproduksi oleh perangkat lunak dan perangkat terdistribusi. Data yang dikirim ke pusat aktivitas dapat diubah dan disimpan menggunakan penyedia analitik real-time atau adaptor batching/penyimpanan. Untuk ringkasan detail Azure Event Hubs, lihat Ringkasan Azure Event Hubs dan fitur Azure Event Hubs.

Panduan cepat ini menjelaskan cara menulis aplikasi Go untuk mengirim peristiwa ke atau menerima peristiwa dari pusat peristiwa.

Nota

Panduan cepat ini didasarkan pada sampel di GitHub di https://github.com/Azure/azure-sdk-for-go/tree/main/sdk/messaging/azeventhubs. Bagian kirim peristiwa didasarkan pada sampel example_producing_events_test.go dan menerimanya didasarkan pada sampel example_processor_test.go . Kode disederhanakan untuk mulai cepat dan semua komentar terperinci dihapus, jadi lihat sampel untuk detail dan penjelasan selengkapnya.

Prasyarat

Untuk menyelesaikan panduan cepat ini, Anda memerlukan persyaratan berikut:

  • Go telah diinstal secara lokal. Ikuti instruksi ini jika perlu.
  • Akun Azure aktif. Jika Anda tidak memiliki langganan Azure, buatlah akun gratis sebelum Anda memulai.
  • Membuat ruang nama Event Hubs dan event hub. Gunakan portal Microsoft Azure untuk membuat namespace jenis Azure Event Hubs, dan mendapatkan info masuk manajemen yang diperlukan aplikasi Anda untuk berkomunikasi dengan hub peristiwa. Untuk membuat namespace layanan dan pusat aktivitas, ikuti prosedur dalam artikel ini.

Kirim acara

Bagian ini memperlihatkan kepada Anda cara membuat aplikasi Go untuk mengirim peristiwa ke pusat aktivitas.

Pasang paket Go

Dapatkan paket Go untuk Azure Event Hubs seperti yang ditunjukkan dalam contoh berikut.

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

Kode untuk mengirim peristiwa ke pusat aktivitas

Berikut adalah kode untuk mengirim peristiwa ke pusat aktivitas. Langkah utama dalam kode adalah:

  1. Buat klien produsen Azure Event Hubs menggunakan string koneksi ke namespace layanan Azure Event Hubs dan nama hub peristiwa.
  2. Buat objek batch dan tambahkan peristiwa sampel ke batch.
  3. Kirim sekumpulan peristiwa ke tujuan yang tepat.

Penting

Ganti NAMESPACE CONNECTION STRING dengan string koneksi ke namespace Event Hubs Anda dan EVENT HUB NAME dengan nama Event Hub dalam kode contoh.

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"),
		},
	}
}

Jangan jalankan aplikasi. Anda harus terlebih dahulu menjalankan aplikasi penerima lalu aplikasi pengirim.

Menerima peristiwa

Membuat akun penyimpanan dan kontainer

Status seperti sewa pada partisi dan titik pemeriksaan dalam peristiwa dibagikan antara penerima menggunakan kontainer Azure Storage. Anda dapat membuat akun penyimpanan dan kontainer dengan Go SDK, tetapi Anda juga dapat membuatnya dengan mengikuti instruksi di Tentang akun penyimpanan Azure.

Ikuti rekomendasi ini saat Anda menggunakan Azure Blob Storage sebagai penyimpanan titik pemeriksaan:

  • Gunakan kontainer terpisah untuk setiap grup konsumen. Anda dapat menggunakan akun penyimpanan yang sama, tetapi menggunakan satu kontainer per setiap grup.
  • Jangan gunakan akun penyimpanan untuk hal lain.
  • Jangan gunakan kontainer untuk hal lain.
  • Buat akun penyimpanan di wilayah yang sama dengan aplikasi yang disebarkan. Jika aplikasi lokal, coba pilih wilayah terdekat yang mungkin.

Pada halaman Akun penyimpanan di portal Azure, di bagian Blob service, pastikan bahwa pengaturan berikut dinonaktifkan.

  • Namespace hierarkis
  • Penghapusan sementara blob
  • Pembuatan Versi

Paket Go

Untuk menerima pesan, dapatkan paket Go untuk Azure Event Hubs seperti yang ditunjukkan dalam contoh berikut.

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

Kode untuk menerima peristiwa dari pusat aktivitas

Berikut adalah kode untuk menerima peristiwa dari pusat aktivitas. Langkah utama dalam kode adalah:

  1. Periksa objek penyimpanan titik pemeriksaan yang mewakili Azure Blob Storage yang digunakan oleh pusat aktivitas untuk titik pemeriksaan.
  2. Buat klien konsumen Azure Event Hubs menggunakan string koneksi ke namespace layanan Azure Event Hubs dan nama hub peristiwa.
  3. Buat pemroses acara menggunakan objek klien dan objek penyimpanan pos pemeriksaan. Prosesor menerima dan memproses peristiwa.
  4. Untuk setiap partisi di pusat aktivitas, buat klien partisi dengan processEvents sebagai fungsi untuk memproses peristiwa.
  5. Jalankan semua klien partisi untuk menerima dan memproses peristiwa.

Penting

Gantilah placeholder berikut dengan nilai aktual:

  • AZURE STORAGE CONNECTION STRING dengan string koneksi untuk akun penyimpanan Azure Anda
  • BLOB CONTAINER NAME dengan nama kontainer blob yang Anda buat di akun penyimpanan
  • NAMESPACE CONNECTION STRING dengan string koneksi ke namespace Event Hubs Anda
  • EVENT HUB NAME dengan nama pusat aktivitas dalam kode sampel.
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())
}

Menjalankan aplikasi penerima dan pengirim

  1. Jalankan aplikasi penerima terlebih dahulu.

  2. Jalankan aplikasi pengirim.

  3. Tunggu sebentar untuk melihat output berikut di jendela penerima.

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

Langkah selanjutnya

Lihat sampel di GitHub di https://github.com/Azure/azure-sdk-for-go/tree/main/sdk/messaging/azeventhubs.

  • Last updated on