Schnellstart: Senden von Ereignissen an oder Empfangen von Ereignissen von Azure Event Hubs

		<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>

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

Zuweisen von Rollen zu Ihrem Microsoft Entra-Benutzer

Stellen Sie bei der lokalen Entwicklung sicher, dass das Benutzerkonto, das eine Verbindung mit Azure Event Hubs herstellt, über die richtigen Berechtigungen verfügt. Sie benötigen die Azure Event Hubs-Rolle "Datenbesitzer ", um Nachrichten zu senden und zu empfangen. Um sich selbst diese Rolle zuzuweisen, benötigen Sie die Rolle des Benutzerzugriffsadministrators oder eine andere Rolle, die die Microsoft.Authorization/roleAssignments/write Aktion enthält. Sie können einem Benutzer Azure RBAC-Rollen über das Azure-Portal, die Azure CLI oder mit Azure PowerShell zuweisen. Weitere Informationen finden Sie auf der Seite Umfang von Azure RBAC verstehen.

Im folgenden Beispiel wird Ihrem Benutzerkonto die Rolle Azure Event Hubs Data Owner zugewiesen. Diese Rolle bietet Vollzugriff auf Azure Event Hubs-Ressourcen. Halten Sie sich in einem echten Szenario an das Prinzip der geringsten Rechte, um Benutzern nur die benötigten Mindestberechtigungen zu erteilen und so die Produktionsumgebung besser zu schützen.

In Azure integrierte Rollen für Azure Event Hubs

Bei Azure Event Hubs ist die Verwaltung der Namespaces und aller zugehörigen Ressourcen über das Azure-Portal und die Azure-Ressourcenverwaltungs-API bereits durch das Azure RBAC-Modell geschützt. Azure bietet die folgenden integrierten Rollen zum Autorisieren des Zugriffs auf einen Event Hubs-Namespace:

• Azure Event Hubs-Datenbesitzer: Ermöglicht den Datenzugriff auf den Event Hubs-Namespace und seine Entitäten (Warteschlangen, Themen, Abonnements und Filter).
• Azure Event Hubs-Datensender: Verwenden Sie diese Rolle, um dem Event Hubs-Namespace und seinen Entitäten Sendezugriff zu erteilen.
• Azure Event Hubs-Datenempfänger: Verwenden Sie diese Rolle, um dem Event Hubs-Namespace und seinen Entitäten Empfangszugriff zu erteilen.

Informationen zum Erstellen einer benutzerdefinierten Rolle finden Sie unter Erforderliche Rechte für Event Hubs-Vorgänge.

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

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

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

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

Verbindungszeichenfolge abrufen

Beim Erstellen eines neuen Namespace wird automatisch eine anfängliche SAS-Richtlinie (Shared Access Signature) mit primären und sekundären Schlüsseln und Verbindungszeichenfolgen generiert, mit denen Sie jeweils die vollständige Kontrolle über sämtliche Aspekte des Namespace haben. Informationen zum Erstellen von Regeln mit eingeschränkteren Rechten für reguläre Absender und Empfänger finden Sie unter Event Hubs-Authentifizierung und Autorisierung.

Ein Client kann die Verbindungszeichenfolge verwenden, um eine Verbindung mit dem Event Hubs-Namespace herzustellen. Führen Sie die folgenden Schritte aus, um die primäre Verbindungszeichenfolge für Ihren Namespace zu kopieren:

1. Wählen Sie auf der Seite Event Hubs-Namespace im linken Menü die Option Freigegebene Zugriffsrichtlinien aus.

2. Wählen Sie auf der Seite Freigegebene Zugriffsrichtlinien die Option RootManageSharedAccessKey aus.

3. Wählen Sie im Fenster Richtlinie: RootManageSharedAccessKey neben Primäre Verbindungszeichenfolge die Schaltfläche „Kopieren“ aus, um die Verbindungszeichenfolge zur späteren Verwendung in die Zwischenablage zu kopieren. Fügen Sie diesen Wert in den Editor oder an einem anderen temporären Speicherort ein.

   

   Auf dieser Seite können Sie den Primärschlüssel, den sekundären Schlüssel, die primäre Verbindungszeichenfolge und die sekundäre Verbindungszeichenfolge kopieren.

Fügen Sie eine Klasse mit dem Namen Senderhinzu, und fügen Sie der Klasse den folgenden Code hinzu:

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();
    }   
}

Fügen Sie eine Klasse mit dem Namen Senderhinzu, und fügen Sie der Klasse den folgenden Code hinzu:

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();
    }
}

Hinzufügen von Code zum Veröffentlichen von Ereignissen im Event Hub

Fügen Sie der publishEvents Klasse eine Methode namens Sender hinzu.

    /**
     * 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();
    }

Stellen Sie bei der lokalen Entwicklung sicher, dass das Benutzerkonto, das auf BLOB-Daten zugreift, über die richtigen Berechtigungen verfügt. Sie benötigen die Berechtigung Mitwirkender an Speicherblobdaten zum Lesen und Schreiben von Blobdaten. Um sich selbst diese Rolle zuzuweisen, müssen Sie der Rolle "Benutzerzugriffsadministrator " oder einer anderen Rolle zugewiesen werden, die die Aktion "Microsoft.Authorization/roleAssignments/write " enthält. Sie können einem Benutzer Azure RBAC-Rollen über das Azure-Portal, die Azure CLI oder mit Azure PowerShell zuweisen. Weitere Informationen finden Sie in der Grundlegendes zum Bereich von Azure RBAC.

In diesem Szenario weisen Sie Ihrem Benutzerkonto Berechtigungen zu, die auf das Speicherkonto festgelegt sind, um dem Prinzip der geringsten Rechte zu folgen. Auf diese Weise erhalten Benutzer nur die erforderlichen Mindestberechtigungen, und es entstehen sicherere Produktionsumgebungen.

Im folgenden Beispiel wird Ihrem Benutzerkonto die Rolle " Storage Blob Data Contributor" zugewiesen, die sowohl Lese- als auch Schreibzugriff auf BLOB-Daten in Ihrem Speicherkonto bietet.

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

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

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

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

Rufen Sie die Verbindungszeichenfolge für das Speicherkonto ab.

Notieren Sie sich die Verbindungszeichenfolge und den Containernamen. Diese Angaben werden im Empfangscode verwendet.

Fügen Sie die folgenden Abhängigkeiten in der datei pom.xml hinzu.

• 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>

Fügen Sie die folgenden Abhängigkeiten in der datei pom.xml hinzu.

• 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>

1. Fügen Sie die folgenden import Anweisungen oben in der Java-Datei hinzu.

   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. Erstellen Sie eine Klasse mit dem Namen Receiver, und fügen Sie der Klasse die folgenden Zeichenfolgenvariablen hinzu. Ersetzen Sie die Platzhalter durch die richtigen Werte.

   Von Bedeutung

   Ersetzen Sie die Platzhalter durch die richtigen Werte.

   • <NAMESPACE NAME> durch den Namen Ihres Event Hubs-Namespace.
   • <EVENT HUB NAME> durch den Namen Ihres Event Hubs im Namespace

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

3. Fügen Sie der Klasse die folgende main Methode hinzu.

   Von Bedeutung

   Ersetzen Sie die Platzhalter durch die richtigen Werte.

   • <STORAGE ACCOUNT NAME> mit dem Namen Ihres Azure Storage-Kontos.
   • <CONTAINER NAME> mit dem Namen des Blob-Containers im Speicherkonto

   // 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");  
   

1. Fügen Sie die folgenden Importanweisungen oben in der Java-Datei hinzu.

   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. Erstellen Sie eine Klasse mit dem Namen Receiver, und fügen Sie der Klasse die folgenden Zeichenfolgenvariablen hinzu. Ersetzen Sie die Platzhalter durch die richtigen Werte.

   Von Bedeutung

   Ersetzen Sie die Platzhalter durch die richtigen Werte.

   • <Event Hubs namespace connection string> durch die Verbindungszeichenfolge für Ihren Event Hubs-Namespace Aktualisierung
   • <Event hub name> durch den Namen Ihres Event Hubs im Namespace
   • <Storage connection string> durch die Verbindungszeichenfolge für Ihr Azure-Speicherkonto
   • <Storage container name> mit dem Namen Ihres Containers in Ihrem Azure Blob Storage.

   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. Fügen Sie der Klasse die folgende main Methode hinzu.

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