Inicio rápido: Envío o recepción de eventos desde 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>

Asignación de roles al usuario de Microsoft Entra

Al desarrollar localmente, asegúrese de que la cuenta de usuario que se conecta a Azure Event Hubs tiene los permisos correctos. Necesita el rol Propietario de datos de Azure Event Hubs para enviar y recibir mensajes. Para asignarse este rol, necesita el rol Administrador de acceso de usuario u otro rol que incluya la Microsoft.Authorization/roleAssignments/write acción. Puede asignar roles RBAC de Azure a un usuario mediante Azure Portal, la CLI de Azure o Azure PowerShell. Para más información, consulte Descripción del ámbito de la página RBAC de Azure .

En el ejemplo siguiente se asigna el rol Azure Event Hubs Data Owner a la cuenta de usuario, que proporciona un acceso completo a los recursos de Azure Event Hubs. En un escenario real, siga el Principio de privilegios mínimos para conceder a los usuarios solo los permisos mínimos necesarios para un entorno de producción más seguro.

Roles integrados de Azure para Azure Event Hubs

En el caso de Azure Event Hubs, la administración de los espacios de nombres y de todos los recursos relacionados mediante Azure Portal y la API de administración de recursos de Azure, ya se ha protegido mediante el modelo de Azure RBAC. Azure proporciona los siguientes roles integrados para autorizar el acceso a un espacio de nombres de Event Hubs:

• Propietario de datos de Azure Event Hubs: Habilita el acceso a datos al espacio de nombres de Event Hubs y a sus entidades (colas, temas, suscripciones y filtros).
• Emisor de datos de Azure Event Hubs: use este rol para proporcionar acceso al emisor a un espacio de nombres de Event Hubs y sus entidades.
• Receptor de datos de Azure Event Hubs: use este rol para proporcionar acceso al receptor a un espacio de nombres de Event Hubs y sus entidades.

Si desea crear un rol personalizado, consulte Derechos necesarios para las operaciones de 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>

Obtén la cadena de conexión

La creación de un espacio de nombres genera automáticamente una directiva de Firma de acceso compartido (SAS) inicial con claves principal y secundaria, y cadenas de conexión en el que ambas conceden control total sobre todos los aspectos del espacio de nombres. Para obtener información sobre cómo crear reglas con derechos más restringidos para remitentes y receptores normales, consulte Autenticación y autorización de Event Hubs.

Un cliente puede usar la cadena de conexión para conectarse al espacio de nombres de Event Hubs. Para copiar la cadena de conexión principal del espacio de nombres, siga estos pasos:

1. En la página Espacio de nombres de Event Hubs, seleccione Directivas de acceso compartido en el menú de la izquierda.

2. En la página Directivas de acceso compartido, seleccione RootManageSharedAccessKey.

3. En la ventana Directiva: RootManageSharedAccessKey, seleccione el botón Copiar que hay junto a Cadena de conexión principal para copiar la cadena de conexión en el Portapapeles para su uso posterior. Pegue este valor en el Bloc de notas o cualquier otra ubicación temporal.

   

   Puede usar esta página para copiar la clave principal, la clave secundaria, la cadena de conexión principal y la cadena de conexión secundaria.

Agregue una clase denominada Sendery agregue el código siguiente a la clase :

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

Agregue una clase denominada Sendery agregue el código siguiente a la clase :

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

Adición de código para publicar eventos en el centro de eventos

Agregue un método denominado publishEvents a la Sender clase :

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

Al desarrollar localmente, asegúrese de que la cuenta de usuario que tiene acceso a los datos de blobs tiene los permisos correctos. Necesita colaborador de datos de Storage Blob para leer y escribir datos de blobs. Para asignarse este rol usted mismo, debe tener asignado el rol Administrador de acceso de usuario, u otro rol que incluya la acción Microsoft.Authorization/roleAssignments/write. Puede asignar roles RBAC de Azure a un usuario mediante Azure Portal, la CLI de Azure o Azure PowerShell. Para obtener más información, vea Descripción del ámbito de Azure RBAC.

En este escenario, asignará permisos a la cuenta de usuario, en el ámbito de la cuenta de almacenamiento, para seguir el principio de privilegios mínimos. Esta práctica solo proporciona a los usuarios los permisos mínimos necesarios y crea entornos de producción más seguros.

En el ejemplo siguiente se asigna el rol Colaborador de datos de Storage Blob a la cuenta de usuario, que proporciona acceso de lectura y escritura a los datos de blobs de la cuenta de almacenamiento.

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>

Obtenga la cadena de conexión a la cuenta de almacenamiento.

Anote la cadena de conexión y el nombre del contenedor. Puede usarlos en el código de recepción.

Agregue las siguientes dependencias en el archivo pom.xml.

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

Agregue las siguientes dependencias en el archivo pom.xml.

• 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. Agregue las siguientes import instrucciones en la parte superior del archivo Java.

   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. Cree una clase denominada Receivery agregue las siguientes variables de cadena a la clase . Reemplace los marcadores de posición por los valores correctos.

   Importante

   Reemplace los marcadores de posición por los valores correctos.

   • <NAMESPACE NAME> por el nombre del espacio de nombres de Event Hubs.
   • <EVENT HUB NAME> con el nombre del centro de eventos en el espacio de nombres.

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

3. Agregue el método siguiente main a la clase .

   Importante

   Reemplace los marcadores de posición por los valores correctos.

   • <STORAGE ACCOUNT NAME> con el nombre de la cuenta de Azure Storage.
   • <CONTAINER NAME> con el nombre del contenedor de blobs de la cuenta de almacenamiento.

   // 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. Agregue las siguientes instrucciones de importación en la parte superior del archivo Java.

   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. Cree una clase denominada Receivery agregue las siguientes variables de cadena a la clase . Reemplace los marcadores de posición por los valores correctos.

   Importante

   Reemplace los marcadores de posición por los valores correctos.

   • <Event Hubs namespace connection string> por la cadena de conexión para el espacio de nombres de Event Hubs. Actualizar
   • <Event hub name> con el nombre del centro de eventos en el espacio de nombres.
   • <Storage connection string> por la cadena de conexión de su cuenta de Azure Storage.
   • <Storage container name> con el nombre del contenedor de 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. Agregue el método siguiente main a la clase .

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