Použití Javy k odesílání nebo příjmu událostí z Azure Event Hubs

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

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

Přiřazení rolí uživateli Microsoft Entra

Při místním vývoji se ujistěte, že uživatelský účet, který se připojuje ke službě Azure Event Hubs, má správná oprávnění. K odesílání a přijímání zpráv budete potřebovat roli vlastníka dat služby Azure Event Hubs. K přiřazení této role budete potřebovat roli Uživatelský přístup Správa istrator nebo jinou roli, která tuto akci zahrnujeMicrosoft.Authorization/roleAssignments/write. Role Azure RBAC můžete uživateli přiřadit pomocí webu Azure Portal, Azure CLI nebo Azure PowerShellu. Další informace o dostupných oborech pro přiřazení rolí najdete na stránce přehledu oboru.

Následující příklad přiřadí roli k vašemu uživatelskému Azure Event Hubs Data Owner účtu, který poskytuje úplný přístup k prostředkům služby Azure Event Hubs. V reálném scénáři postupujte podle principu nejnižšího oprávnění , aby uživatelům poskytla pouze minimální oprávnění potřebná pro bezpečnější produkční prostředí.

Předdefinované role Azure pro Azure Event Hubs

Pro Službu Azure Event Hubs je správa oborů názvů a všech souvisejících prostředků prostřednictvím webu Azure Portal a rozhraní API pro správu prostředků Azure již chráněno pomocí modelu Azure RBAC. Azure poskytuje následující předdefinované role Azure pro autorizaci přístupu k oboru názvů služby Event Hubs:

• Vlastník dat služby Azure Event Hubs: Umožňuje přístup k datům k oboru názvů služby Event Hubs a jeho entitám (fronty, témata, odběry a filtry).
• Odesílatel dat služby Azure Event Hubs: Pomocí této role dáte odesílateli přístup k oboru názvů služby Event Hubs a jeho entitám.
• Příjemce dat služby Azure Event Hubs: Pomocí této role dejte příjemci přístup k oboru názvů služby Event Hubs a jeho entitám.

Pokud chcete vytvořit vlastní roli, přečtěte si téma Práva potřebná pro operace služby Event Hubs.

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>

Získání připojovacího řetězce

Vytvoření nového oboru názvů automaticky vygeneruje počáteční zásadu sdíleného přístupového podpisu (SAS) s primárními a sekundárními klíči a připojovací řetězec, které každý z nich uděluje úplnou kontrolu nad všemi aspekty oboru názvů. Informace o vytváření pravidel s více omezenými právy pro běžné odesílatele a příjemce najdete v tématu Ověřování a autorizace služby Event Hubs.

Klient se může pomocí připojovací řetězec připojit k oboru názvů služby Event Hubs. Pokud chcete zkopírovat primární připojovací řetězec pro váš obor názvů, postupujte takto:

1. Na stránce Obor názvů centra událostí vyberte v nabídce vlevo zásady sdíleného přístupu.

2. Na stránce Zásady sdíleného přístupu vyberte RootManageSharedAccessKey.

3. V okně Zásady: RootManageSharedAccessKey vyberte tlačítko kopírování vedle primárního Připojení ion String a zkopírujte připojovací řetězec do schránky pro pozdější použití. Vložte tuto hodnotu do Poznámkového bloku nebo jiného dočasného umístění.

   

   Na této stránce můžete kopírovat primární klíč, sekundární klíč, primární připojovací řetězec a sekundární připojovací řetězec.

Přidejte třídu s názvem Sendera přidejte do třídy následující kód:

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

Přidejte třídu s názvem Sendera přidejte do třídy následující kód:

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

Přidání kódu pro publikování událostí do centra událostí

Přidejte metodu Sender pojmenovanou publishEvents do třídy:

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

Při místním vývoji se ujistěte, že uživatelský účet, který přistupuje k datům objektů blob, má správná oprávnění. K čtení a zápisu dat objektů blob budete potřebovat Přispěvatel dat objektů blob služby Storage. Abyste mohli tuto roli přiřadit sami sobě, musíte mít přiřazenou roli User Access Správa istrator nebo jinou roli, která zahrnuje akci Microsoft.Authorization/roleAssignments/write. Role Azure RBAC můžete uživateli přiřadit pomocí webu Azure Portal, Azure CLI nebo Azure PowerShellu. Další informace o dostupných oborech pro přiřazení rolí najdete na stránce přehledu oboru.

V tomto scénáři přiřadíte oprávnění k vašemu uživatelskému účtu s vymezeným oborem účtu úložiště, abyste postupovali podle zásady nejnižších oprávnění. Tento postup poskytuje uživatelům jenom minimální potřebná oprávnění a vytváří bezpečnější produkční prostředí.

Následující příklad přiřadí roli Přispěvatel dat v objektech blob služby Storage k vašemu uživatelskému účtu, který poskytuje přístup ke čtení i zápisu k datům objektů blob v účtu úložiště.

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>

Získejte připojovací řetězec k účtu úložiště.

Poznamenejte si připojovací řetězec a název kontejneru. Použijete je v příjmovém kódu.

Do souboru pom.xml přidejte následující závislosti.

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

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

Do souboru pom.xml přidejte následující závislosti.

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

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

1. Na začátek souboru Java přidejte následující import příkazy.

   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. Vytvořte třídu s názvem Receivera přidejte do třídy následující řetězcové proměnné. Zástupné symboly nahraďte správnými hodnotami.

   Důležité

   Zástupné symboly nahraďte správnými hodnotami.

   • <NAMESPACE NAME> nahraďte názvem vašeho oboru názvů Event Hubs.
   • <EVENT HUB NAME> s názvem vašeho centra událostí v oboru názvů.

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

3. Do třídy přidejte následující main metodu.

   Důležité

   Zástupné symboly nahraďte správnými hodnotami.

   • <STORAGE ACCOUNT NAME> s názvem vašeho účtu Azure Storage.
   • <CONTAINER NAME> s názvem kontejneru objektů blob v účtu úložiště

   // 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. Na začátek souboru Java přidejte následující příkazy importu .

   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. Vytvořte třídu s názvem Receivera přidejte do třídy následující řetězcové proměnné. Zástupné symboly nahraďte správnými hodnotami.

   Důležité

   Zástupné symboly nahraďte správnými hodnotami.

   • <Event Hubs namespace connection string>se připojovací řetězec do oboru názvů služby Event Hubs. Aktualizovat
   • <Event hub name> s názvem vašeho centra událostí v oboru názvů.
   • <Storage connection string>pomocí připojovací řetězec k vašemu účtu úložiště Azure.
   • <Storage container name> s názvem kontejneru ve službě 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. Do třídy přidejte následující main metodu.

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