Hızlı Başlangıç: Azure Event Hubs'a olay gönderme veya olay alma

Bu Hızlı Başlangıçta, azure-messaging-eventhubs Java paketini kullanarak azure olay hub'ına olay göndermeyi ve azure olay hub'ından olay almayı öğreneceksiniz.

Tavsiye

Bir Spring uygulamasında Azure Event Hubs kaynaklarıyla çalışıyorsanız Spring Cloud Azure'ı alternatif olarak değerlendirmenizi öneririz. Spring Cloud Azure, Azure hizmetleriyle sorunsuz Spring tümleştirmesi sağlayan açık kaynak bir projedir. Spring Cloud Azure hakkında daha fazla bilgi edinmek ve Event Hubs kullanarak bir örnek görmek için bkz. Azure Event Hubs ile Spring Cloud Stream.

Önkoşullar

Azure Event Hubs'da yeniyseniz bu hızlı başlangıcı gerçekleştirmeden önce event hubs'a genel bakış konusuna bakın.

Bu hızlı başlangıcı tamamlamak için aşağıdaki önkoşullara ihtiyacınız vardır:

• Microsoft Azure aboneliği. Azure Event Hubs da dahil olmak üzere Azure hizmetlerini kullanmak için bir aboneliğe ihtiyacınız vardır. Mevcut bir Azure hesabınız yoksa ücretsiz deneme sürümüne kaydolabilir veya hesap oluştururken MSDN abone avantajlarınızı kullanabilirsiniz.
• Java geliştirme ortamı. Bu hızlı başlangıçta Eclipse kullanılır. Sürüm 8 veya üzeri olan Java Development Kit (JDK) gereklidir.
• Event Hubs ad alanı ve bir olay hub'ı oluşturun. İlk adım, Event Hubs türünde bir ad alanı oluşturmak ve uygulamanızın olay hub'ı ile iletişim kurmak için ihtiyaç duyduğu yönetim kimlik bilgilerini almak için Azure portalını kullanmaktır. Ad alanı ve olay hub'ı oluşturmak için bu makaledeki yordamı izleyin. Ardından, Event Hubs ad alanının bağlantı dizesini almak içinmakale yönergelerini takip edin: Bağlantı dizesini alma. Bağlantı dizesini bu hızlı başlangıcın ilerleyen bölümlerinde kullanacaksınız.

Olayları gönderme

Bu bölümde, olayları bir olay hub'ına göndermek için nasıl Java uygulaması oluşturulacağı gösterilmektedir.

Azure Event Hubs kitaplığına referans ekleme

İlk olarak, sık kullandığınız Java geliştirme ortamında bir konsol/kabuk uygulaması için yeni bir Maven projesi oluşturun. Dosyayı aşağıdaki gibi güncelleştirin pom.xml . Event Hubs için Java istemci kitaplığı Maven Central Repository'de kullanılabilir.

Parolasız (Önerilen)

		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.20.2</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-identity</artifactId>
		    <version>1.16.1</version>
		    <scope>compile</scope>
		</dependency>

Bağlantı Dizesi

		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.20.2</version>
		</dependency>

Uyarı

Sürümü Maven deposunda yayımlanan en son sürüme güncelleştirin.

Azure'da uygulamanın kimliğini doğrulama

Bu hızlı başlangıçta Azure Event Hubs'a bağlanmanın iki yolu gösterilmektedir:

• Parolasız. Bir Event Hubs ad alanına bağlanmak için Microsoft Entra Id ve rol tabanlı erişim denetiminde (RBAC) güvenlik sorumlunuzu kullanın. Kodunuzda, yapılandırma dosyasında veya Azure Key Vault gibi güvenli depolama alanında sabit kodlanmış bağlantı dizelerine sahip olma konusunda endişelenmeniz gerekmez.
• Bağlantı dizesi. Event Hubs ad alanına bağlanmak için bağlantı dizesi kullanın. Azure'da yeniyseniz bağlantı dizesi seçeneğini daha kolay takip edebilirsiniz.

Gerçek dünyadaki uygulamalarda ve üretim ortamlarında parolasız seçeneği kullanmanızı öneririz. Daha fazla bilgi için bkz. Azure hizmetleri için Service Bus kimlik doğrulaması ve yetkilendirmesi ve Parolasız bağlantılar.

Parolasız (Önerilen)

Microsoft Entra kullanıcınıza rol atama

Yerel olarak geliştirirken, Azure Event Hubs'a bağlanan kullanıcı hesabının doğru izinlere sahip olduğundan emin olun. İleti göndermek ve almak için Azure Event Hubs Veri Sahibi rolüne ihtiyacınız vardır. Kendinize bu rolü atamak için Kullanıcı Erişimi Yöneticisi rolüne veya eylemi içeren Microsoft.Authorization/roleAssignments/write başka bir role ihtiyacınız vardır. Azure portalı, Azure CLI veya Azure PowerShell'i kullanarak kullanıcıya Azure RBAC rolleri atayabilirsiniz. Daha fazla bilgi için bkz. Azure RBAC kapsamını anlama sayfası.

Aşağıdaki örnekte rol, Azure Event Hubs kaynaklarına tam erişim sağlayan kullanıcı hesabınıza atanır Azure Event Hubs Data Owner . Gerçek bir senaryoda, kullanıcılara yalnızca daha güvenli bir üretim ortamı için gereken minimum izinleri vermek için En Az Ayrıcalık İlkesi'ni izleyin.

Azure Event Hubs için Azure yerleşik rolleri

Azure Event Hubs için, Azure portalı ve Azure kaynak yönetimi API'sini kullanarak ad alanlarının ve tüm ilgili kaynakların yönetimi Azure RBAC modeli kullanılarak zaten korunur. Azure, Event Hubs ad alanına erişim yetkisi vermek için aşağıdaki yerleşik rolleri sağlar:

• Azure Event Hubs Veri Sahibi: Event Hubs ad alanına ve varlıklarına (kuyruklar, konular, abonelikler ve filtreler) veri erişimini etkinleştirir.
• Azure Event Hubs Veri Göndereni: Gönderene Event Hubs ad alanına ve varlıklarına erişim vermek için bu rolü kullanın.
• Azure Event Hubs Veri Alıcısı: Alıcıya Event Hubs ad alanına ve varlıklarına erişim vermek için bu rolü kullanın.

Özel bir rol oluşturmak istiyorsanız, Event Hubs işlemleri için gereken haklar bölümüne bakın.

Önemli

Çoğu durumda rol atamasının Azure'a yayılması bir veya iki dakika sürer. Nadir durumlarda, sekiz dakikaya kadar sürebilir. Kodunuzu ilk kez çalıştırdığınızda kimlik doğrulama hataları alıyorsanız, birkaç dakika bekleyin ve yeniden deneyin.

Azure portalı

1. Azure portalında ana arama çubuğunu veya sol gezintiyi kullanarak Event Hubs ad alanınızı bulun.

2. Genel bakış sayfasında, sol taraftaki menüden Erişim denetimi (IAM) öğesini seçin.

3. Erişim denetimi (IAM) sayfasında Rol atamaları sekmesini seçin.

4. Üst menüden + Ekle'yi seçin. Ardından Rol ataması ekle'yi seçin.

   

5. Sonuçları istenen role göre filtrelemek için arama kutusunu kullanın. Bu örnek için eşleşen sonucu arayın Azure Event Hubs Data Owner ve seçin. Ardından İleri'yi seçin.

6. Erişim ata'nın altında Kullanıcı, grup veya hizmet sorumlusu'na tıklayın. Ardından + Üye seç'i seçin.

7. İletişim kutusunda Microsoft Entra kullanıcı adınızı (genellikle user@domain e-posta adresiniz) arayın. İletişim kutusunun alt kısmındaki Seç'i seçin.

8. Son sayfaya gitmek için Gözden geçir ve ata'yı seçin. İşlemi tamamlamak için Gözden geçir ve yeniden ata komutunu seçin.

Azure CLI

Azure CLI kullanarak kaynak düzeyinde bir rol atamak için önce komutunu kullanarak az eventhubs namespace show kaynak kimliğini alın. parametresini kullanarak --query çıkış özelliklerini filtreleyebilirsiniz.

az eventhubs namespace show -g '<your-event-hub-resource-group>' -n '<your-event-hub-name>' --query id

Önceki komuttan çıktıyı Id kopyalayın. Ardından az role komutunu kullanarak roller atayabilirsiniz.

az role assignment create --assignee "<user@domain>" \
--role "Azure Event Hubs Data Owner" \
--scope "<your-resource-id>"

PowerShell

Azure PowerShell kullanarak kaynak düzeyinde bir rol atamak için önce komutunu kullanarak Get-AzResource kaynak kimliğini alın.

Get-AzResource -ResourceGroupName "<your-event-hub-resource-group>" -Name "<your-event-hub-name>"

Id Önceki komut çıkışındaki değeri kopyalayın. Ardından New-AzRoleAssignment komutunu kullanarak roller atayabilirsiniz.

New-AzRoleAssignment -SignInName <user@domain> `
-RoleDefinitionName "Azure Event Hubs Data Owner" `
-Scope <yourResourceId>

Bağlantı Dizesi

Bağlantı dizesini al

Yeni bir ad alanı oluşturmak, birincil ve ikincil anahtarlar ve her birinin ad alanının tüm yönleri üzerinde tam denetim sağlayan bağlantı dizesi ile otomatik olarak bir ilk Paylaşılan Erişim İmzası (SAS) ilkesi oluşturur. Normal gönderenler ve alıcılar için daha kısıtlı haklara sahip kurallar oluşturma hakkında bilgi için bkz. Event Hubs kimlik doğrulaması ve yetkilendirmesi.

İstemci, Event Hubs ad alanına bağlanmak için bağlantı dizesi kullanabilir. Ad alanınızın birincil bağlantı dizesi kopyalamak için şu adımları izleyin:

1. Olay Hub Ad Alanı sayfasında, soldaki menüden Paylaşılan erişim ilkeleri'ni seçin.

2. Paylaşılan erişim ilkeleri sayfasında RootManageSharedAccessKey'i seçin.

3. İlke: RootManageSharedAccessKey penceresinde, Birincil Bağlantı Dizesi'nin yanındaki kopyala düğmesine basarak bağlantı dizesini daha sonra kullanmak üzere panonuza kopyalayın. Bu değeri Not Defteri veya başka bir geçici konuma yapıştırın.

   

   Birincil anahtar, ikincil anahtar, birincil bağlantı dizesi ve ikincil bağlantı dizesi kopyalamak için bu sayfayı kullanabilirsiniz.

Olay hub'ına ileti göndermek için kod yazma

Parolasız (Önerilen)

adlı Senderbir sınıf ekleyin ve sınıfına aşağıdaki kodu ekleyin:

Önemli

• Olay Hub'ları ad alanınızın adını <NAMESPACE NAME> ile güncelleştirin.
• Etkinlik hub'ınızın adıyla <EVENT HUB NAME>'yi güncelleyin.

package ehubquickstart;

import com.azure.messaging.eventhubs.*;
import java.util.Arrays;
import java.util.List;

import com.azure.identity.*;

public class SenderAAD {

    // replace <NAMESPACE NAME> with the name of your Event Hubs namespace.
    // Example: private static final String namespaceName = "contosons.servicebus.windows.net";
    private static final String namespaceName = "<NAMESPACE NAME>.servicebus.windows.net";

    // Replace <EVENT HUB NAME> with the name of your event hub. 
    // Example: private static final String eventHubName = "ordersehub";
    private static final String eventHubName = "<EVENT HUB NAME>";

    public static void main(String[] args) {
        publishEvents();
    }
    /**
     * Code sample for publishing events.
     * @throws IllegalArgumentException if the EventData is bigger than the max batch size.
     */
    public static void publishEvents() {
        // create a token using the default Azure credential        
        DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
                .authorityHost(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD)
                .build();

        // create a producer client        
        EventHubProducerClient producer = new EventHubClientBuilder()        
            .fullyQualifiedNamespace(namespaceName)
            .eventHubName(eventHubName)
            .credential(credential)
            .buildProducerClient();

        // sample events in an array
        List<EventData> allEvents = Arrays.asList(new EventData("Foo"), new EventData("Bar"));

        // create a batch
        EventDataBatch eventDataBatch = producer.createBatch();

        for (EventData eventData : allEvents) {
            // try to add the event from the array to the batch
            if (!eventDataBatch.tryAdd(eventData)) {
                // if the batch is full, send it and then create a new batch
                producer.send(eventDataBatch);
                eventDataBatch = producer.createBatch();

                // Try to add that event that couldn't fit before.
                if (!eventDataBatch.tryAdd(eventData)) {
                    throw new IllegalArgumentException("Event is too large for an empty batch. Max size: "
                        + eventDataBatch.getMaxSizeInBytes());
                }
            }
        }
        // send the last batch of remaining events
        if (eventDataBatch.getCount() > 0) {
            producer.send(eventDataBatch);
        }
        producer.close();
    }   
}

Bağlantı Dizesi

adlı Senderbir sınıf ekleyin ve sınıfına aşağıdaki kodu ekleyin:

Önemli

Event Hubs ad alanınıza bağlantı dizesini <Event Hubs namespace connection string> güncelleştirin. Ad alanında yer alan etkinlik hub'ının adıyla <Event hub name>'yi güncelle.

import com.azure.messaging.eventhubs.*;
import java.util.Arrays;
import java.util.List;

public class Sender {
    private static final String connectionString = "<Event Hubs namespace connection string>";
    private static final String eventHubName = "<Event hub name>";

    public static void main(String[] args) {
        publishEvents();
    }
}

Olay hub'ına olay yayımlamak için kod ekleme

sınıfına publishEvents adlı Sender bir yöntem ekleyin:

    /**
     * Code sample for publishing events.
     * @throws IllegalArgumentException if the EventData is bigger than the max batch size.
     */
    public static void publishEvents() {
        // create a producer client
        EventHubProducerClient producer = new EventHubClientBuilder()
            .connectionString(connectionString, eventHubName)
            .buildProducerClient();

        // sample events in an array
        List<EventData> allEvents = Arrays.asList(new EventData("Foo"), new EventData("Bar"));

        // create a batch
        EventDataBatch eventDataBatch = producer.createBatch();

        for (EventData eventData : allEvents) {
            // try to add the event from the array to the batch
            if (!eventDataBatch.tryAdd(eventData)) {
                // if the batch is full, send it and then create a new batch
                producer.send(eventDataBatch);
                eventDataBatch = producer.createBatch();

                // Try to add that event that couldn't fit before.
                if (!eventDataBatch.tryAdd(eventData)) {
                    throw new IllegalArgumentException("Event is too large for an empty batch. Max size: "
                        + eventDataBatch.getMaxSizeInBytes());
                }
            }
        }
        // send the last batch of remaining events
        if (eventDataBatch.getCount() > 0) {
            producer.send(eventDataBatch);
        }
        producer.close();
    }

Programı oluşturun ve hata olmadığından emin olun. Alıcı programını çalıştırdıktan sonra bu programı çalıştıracaksınız.

Olayları al

Bu öğreticideki kod, GitHub'daki EventProcessorClient örneğini temel alır ve tam çalışan uygulamayı görmek için inceleyebilirsiniz.

Azure Blob Depolama'yı denetim noktası deposu olarak kullanırken şu önerileri izleyin:

• Her tüketici grubu için ayrı bir kapsayıcı kullanın. Aynı depolama hesabını kullanabilirsiniz, ancak her grup için bir kapsayıcı kullanabilirsiniz.
• Depolama hesabını başka bir şey için kullanmayın.
• Kapsayıcıyı başka hiçbir şey için kullanmayın.
• Depolama hesabını dağıtılan uygulamayla aynı bölgede oluşturun. Uygulama şirket içindeyse, mümkün olan en yakın bölgeyi seçmeyi deneyin.

Azure portalının Depolama hesabı sayfasındaki Blob hizmeti bölümünde aşağıdaki ayarların devre dışı bırakıldığından emin olun.

• Hiyerarşik ad alanı
• Blob geçici silme
• Sürüm oluşturma

Azure Depolama Alanı ve blob kapsayıcısı oluşturun

Bu hızlı başlangıçta denetim noktası deposu olarak Azure Depolama'yı (özellikle Blob Depolama) kullanacaksınız. Denetim noktası oluşturma, bir olay işlemcisinin bir bölümde başarıyla işlenen son olayın konumunu işaretlediği veya kaydettiği bir işlemdir. Bir denetim noktasını işaretleme işlemi genellikle olayları işleyen işlev içinde yapılır. Denetim noktası oluşturma hakkında daha fazla bilgi edinmek için bkz. Olay işlemcisi.

Azure Depolama hesabı oluşturmak için bu adımları izleyin.

1. Azure Depolama hesabı oluşturma
2. Blob kapsayıcısı oluştur
3. Blob kapsayıcısı için kimlik doğrulaması

Parolasız (Önerilen)

Yerel olarak geliştirirken blob verilerine erişen kullanıcı hesabının doğru izinlere sahip olduğundan emin olun. Blob verilerini okumak ve yazmak için Depolama Blob Verisi Katkıcısı rolüne ihtiyacınız var. Kendinize bu rolü atamak için Kullanıcı Erişimi Yöneticisi rolüne veya Microsoft.Authorization/roleAssignments/write eylemini içeren başka bir role atanmalısınız. Azure portalı, Azure CLI veya Azure PowerShell'i kullanarak kullanıcıya Azure RBAC rolleri atayabilirsiniz. Daha fazla bilgi için Azure RBAC kapsamını anlama bölümüne bakın.

Bu senaryoda, En Az Ayrıcalık İlkesi'ni izlemek için depolama hesabı kapsamındaki kullanıcı hesabınıza izinler atarsınız. Bu uygulama kullanıcılara yalnızca gereken minimum izinleri verir ve daha güvenli üretim ortamları oluşturur.

Aşağıdaki örnek, depolama hesabınızdaki blob verilerine hem okuma hem de yazma erişimi sağlayan Depolama Blob Verileri Katkıda Bulunanı rolünü kullanıcı hesabınıza atar.

Önemli

Çoğu durumda rol atamasının Azure'a yayılması bir veya iki dakika sürer. Nadir durumlarda, sekiz dakikaya kadar sürebilir. Kodunuzu ilk kez çalıştırdığınızda kimlik doğrulama hataları alıyorsanız, birkaç dakika bekleyin ve yeniden deneyin.

Azure portalı

1. Azure portalında ana arama çubuğunu veya sol gezintiyi kullanarak depolama hesabınızı bulun.

2. Depolama hesabı sayfasında sol taraftaki menüden Erişim denetimi (IAM) öğesini seçin.

3. Erişim denetimi (IAM) sayfasında Rol atamaları sekmesini seçin.

4. Üst menüden + Ekle'yi seçin. Ardından Rol ataması ekle'yi seçin.

   

5. Sonuçları istenen role göre filtrelemek için arama kutusunu kullanın. Bu örnek için Depolama Blobu Veri Katkıda Bulunanını arayın. Eşleşen sonucu seçin ve ardından Sonrakiseçin.

6. Erişim atama altında, Kullanıcı, grup veya hizmet sorumlusu seçeneğini seçin ve ardından + Üyeleri seç seçeneğine tıklayın.

7. İletişim kutusunda Microsoft Entra kullanıcı adınızı (genellikle user@domain e-posta adresiniz) arayın ve iletişim kutusunun alt kısmındaki Seç'i seçin.

8. Son sayfaya gitmek için Gözden geçir ve ata'yı seçin. İşlemi tamamlamak için Gözden geçir ve yeniden ata komutunu seçin.

Azure CLI

Azure CLI kullanarak kaynak düzeyinde bir rol atamak için önce komutunu kullanarak az storage account show kaynak kimliğini alın. parametresini kullanarak --query çıkış özelliklerini filtreleyebilirsiniz.

az storage account show --resource-group '<your-resource-group-name>' --name '<your-storage-account-name>' --query id

Önceki komuttan çıktıyı Id kopyalayın. Ardından az role komutunu kullanarak roller atayabilirsiniz.

az role assignment create --assignee "<user@domain>" \
    --role "Storage Blob Data Contributor" \
    --scope "<your-resource-id>"

PowerShell

Azure PowerShell kullanarak kaynak düzeyinde bir rol atamak için önce komutunu kullanarak Get-AzResource kaynak kimliğini alın.

Get-AzResource -ResourceGroupName "<yourResourceGroupname>" -Name "<yourStorageAccountName>"

Id Önceki komut çıkışındaki değeri kopyalayın. Ardından New-AzRoleAssignment komutunu kullanarak roller atayabilirsiniz.

New-AzRoleAssignment -SignInName <user@domain> `
    -RoleDefinitionName "Storage Blob Data Contributor" `
    -Scope <yourStorageAccountId>

Bağlantı Dizesi

Depolama hesabına bağlantı dizesini alın.

Bağlantı dizesini ve kapsayıcı adını not edin. Bunları alım kodunda kullanırsınız.

Java projenize Event Hubs kitaplıkları ekleme

Parolasız (Önerilen)

pom.xml dosyasına aşağıdaki bağımlılıkları ekleyin.

• azure-messaging-eventhubs
• azure-messaging-eventhubs-checkpointstore-blob
• azure-identity

	<dependencies>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.20.2</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
		    <version>1.20.6</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-identity</artifactId>
		    <version>1.16.1</version>
		    <scope>compile</scope>
		</dependency>	
	</dependencies>

Bağlantı Dizesi

pom.xml dosyasına aşağıdaki bağımlılıkları ekleyin.

• azure-messaging-eventhubs
• azure-messaging-eventhubs-checkpointstore-blob

	<dependencies>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.20.2</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
		    <version>1.20.6</version>
		</dependency>
	</dependencies>

Parolasız (Önerilen)

1. Java dosyasının en üstüne aşağıdaki import deyimleri ekleyin.

   import com.azure.messaging.eventhubs.*;
   import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
   import com.azure.messaging.eventhubs.models.*;
   import com.azure.storage.blob.*;
   import java.util.function.Consumer;
   
   import com.azure.identity.*;
   

2. adlı Receiverbir sınıf oluşturun ve sınıfına aşağıdaki dize değişkenlerini ekleyin. Yer tutucuları doğru değerlerle değiştirin.

   Önemli

   Yer tutucuları doğru değerlerle değiştirin.

   • Event Hubs ad alanınızın adıyla <NAMESPACE NAME>'ı değiştirin.
   • <EVENT HUB NAME> ad alanında olay hub'ınızın adını yazın.

   private static final String namespaceName = "<NAMESPACE NAME>.servicebus.windows.net";
   private static final String eventHubName = "<EVENT HUB NAME>";
   

3. Sınıfına aşağıdaki main yöntemi ekleyin.

   Önemli

   Yer tutucuları doğru değerlerle değiştirin.

   • <STORAGE ACCOUNT NAME> azure depolama hesabınızın adıyla birlikte.
   • <CONTAINER NAME> depolama hesabındaki blob kapsayıcısının adıyla

   // create a token using the default Azure credential
   DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
           .authorityHost(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD)
           .build();
   
   // Create a blob container client that you use later to build an event processor client to receive and process events
   BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
           .credential(credential)
           .endpoint("https://<STORAGE ACCOUNT NAME>.blob.core.windows.net")
           .containerName("<CONTAINER NAME>")
           .buildAsyncClient();
   
   // Create an event processor client to receive and process events and errors.
   EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
       .fullyQualifiedNamespace(namespaceName)
       .eventHubName(eventHubName)
       .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
       .processEvent(PARTITION_PROCESSOR)
       .processError(ERROR_HANDLER)
       .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))            
       .credential(credential)
       .buildEventProcessorClient();
   
   System.out.println("Starting event processor");
   eventProcessorClient.start();
   
   System.out.println("Press enter to stop.");
   System.in.read();
   
   System.out.println("Stopping event processor");
   eventProcessorClient.stop();
   System.out.println("Event processor stopped.");
   
   System.out.println("Exiting process");  
   

Bağlantı Dizesi

1. Java dosyasının en üstüne aşağıdaki içeri aktarma deyimlerini ekleyin.

   import com.azure.messaging.eventhubs.*;
   import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
   import com.azure.messaging.eventhubs.models.*;
   import com.azure.storage.blob.*;
   import java.util.function.Consumer;
   

2. adlı Receiverbir sınıf oluşturun ve sınıfına aşağıdaki dize değişkenlerini ekleyin. Yer tutucuları doğru değerlerle değiştirin.

   Önemli

   Yer tutucuları doğru değerlerle değiştirin.

   • Event Hubs ad alanınıza bağlanmak için bağlantı dizesini <Event Hubs namespace connection string> ile kullanın. Güncelleştir
   • <Event hub name> ad alanında olay hub'ınızın adını yazın.
   • <Storage connection string> Azure depolama hesabınıza ait bağlantı dizesiyle.
   • <Storage container name> Azure Blob depolama alanınızdaki kapsayıcınızın adıyla.

   private static final String connectionString = "<Event Hubs namespace connection string>";
   private static final String eventHubName = "<Event hub name>";
   private static final String storageConnectionString = "<Storage connection string>";
   private static final String storageContainerName = "<Storage container name>";
   

3. Sınıfına aşağıdaki main yöntemi ekleyin.

   public static void main(String[] args) throws Exception {
       // Create a blob container client that you use later to build an event processor client to receive and process events
       BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
           .connectionString(storageConnectionString)
           .containerName(storageContainerName)
           .buildAsyncClient();
   
       // Create a builder object that you will use later to build an event processor client to receive and process events and errors.
       EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
           .connectionString(connectionString, eventHubName)
           .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
           .processEvent(PARTITION_PROCESSOR)
           .processError(ERROR_HANDLER)
           .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient));
   
       // Use the builder object to create an event processor client
       EventProcessorClient eventProcessorClient = eventProcessorClientBuilder.buildEventProcessorClient();
   
       System.out.println("Starting event processor");
       eventProcessorClient.start();
   
       System.out.println("Press enter to stop.");
       System.in.read();
   
       System.out.println("Stopping event processor");
       eventProcessorClient.stop();
       System.out.println("Event processor stopped.");
   
       System.out.println("Exiting process");
   }
   

4. Olayları ve hataları işleyen iki yardımcı yöntemi (PARTITION_PROCESSOR ve ERROR_HANDLER) sınıfına Receiver ekleyin.

   public static final Consumer<EventContext> PARTITION_PROCESSOR = eventContext -> {
       PartitionContext partitionContext = eventContext.getPartitionContext();
       EventData eventData = eventContext.getEventData();
   
       System.out.printf("Processing event from partition %s with sequence number %d with body: %s%n",
           partitionContext.getPartitionId(), eventData.getSequenceNumber(), eventData.getBodyAsString());
   
       // Every 10 events received, it will update the checkpoint stored in Azure Blob Storage.
       if (eventData.getSequenceNumber() % 10 == 0) {
           eventContext.updateCheckpoint();
       }
   };
   
   public static final Consumer<ErrorContext> ERROR_HANDLER = errorContext -> {
       System.out.printf("Error occurred in partition processor for partition %s, %s.%n",
           errorContext.getPartitionContext().getPartitionId(),
           errorContext.getThrowable());
   };
   

5. Programı oluşturun ve hata olmadığından emin olun.

Uygulamaları çalıştırma

1. Önce Alıcı uygulamasını çalıştırın.

2. Ardından Sender uygulamasını çalıştırın.

3. Alıcı uygulaması penceresinde, Gönderen uygulaması tarafından yayımlanan olayları gördüğünüzden emin olun.

   Starting event processor
   Press enter to stop.
   Processing event from partition 0 with sequence number 331 with body: Foo
   Processing event from partition 0 with sequence number 332 with body: Bar
   

4. Uygulamayı durdurmak için alıcı uygulama penceresinde ENTER tuşuna basın.

   Starting event processor
   Press enter to stop.
   Processing event from partition 0 with sequence number 331 with body: Foo
   Processing event from partition 0 with sequence number 332 with body: Bar
   
   Stopping event processor
   Event processor stopped.
   Exiting process
   

İlgili içerik

GitHub'da aşağıdaki örneklere bakın:

• azure-messaging-eventhubs örnekleri
• azure-messaging-eventhubs-checkpointstore-blob örnekleri.