Usar o Java para enviar eventos ou receber eventos dos Hubs de Eventos do Azure

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

Atribuir funções ao usuário do Microsoft Entra

No caso de desenvolvimento local, verifique se a conta de usuário que se conecta aos Hubs de Eventos do Azure tem as permissões corretas. Você precisará da função Proprietário de Dados dos Hubs de Eventos do Azure para enviar e receber mensagens. Para atribuir essa função a si mesmo, você precisará da função Administrador de acesso do usuário ou de outra função que inclua a ação Microsoft.Authorization/roleAssignments/write. É possível atribuir funções RBAC do Azure a um usuário usando o portal do Azure, a CLI do Azure ou o Azure PowerShell. Saiba mais sobre os escopos disponíveis para atribuições de função na página de visão geral do escopo.

O exemplo a seguir atribui a função Azure Event Hubs Data Owner à sua conta de usuário, que fornece acesso completo aos recursos dos Hubs de Eventos do Azure. Em um cenário real, siga o Princípio do Privilégio Mínimo para dar aos usuários apenas as permissões mínimas necessárias para um ambiente de produção mais seguro.

Funções interna do Azure par Hubs de Eventos do Azure

Para os Hubs de Eventos do Azure, o gerenciamento de namespaces e de todos os recursos relacionados por meio do portal do Azure e da API de gerenciamento de recursos do Azure já está protegido pelo modelo Azure RBAC. O Azure fornece as funções internas do Azure abaixo para autorizar o acesso a um namespace dos Hubs de Eventos:

• Proprietário de Dados dos Hubs de Eventos do Azure: permite o acesso a dados no namespace dos Hubs de Eventos e às respectivas entidades (filas, tópicos, assinaturas e filtros)
• Remetente de Dados dos Hubs de Eventos do Azure: use essa função para fornecer acesso de remetente ao namespace dos Hubs de Eventos e às respectivas entidades.
• Destinatário de Dados dos Hubs de Eventos do Azure: use essa função para fornecer acesso de remetente ao namespace dos Hubs de Eventos e às respectivas entidades.

Se você quiser criar uma função personalizada, confira Direitos exigidos para operações dos Hubs de Eventos.

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>

Obtenha a cadeia de conexão

A criação de um namespace gera automaticamente uma política inicial de SAS (Assinatura de Acesso Compartilhado) com chaves e cadeias de conexão primárias e secundárias que concedem controle completo sobre todos os aspectos do namespace. Confira Autenticação e autorização dos Hubs de Eventos para obter informações de como criar regras com direitos mais restritos para remetentes e destinatários regulares.

Um cliente pode usar a cadeia de conexão para se conectar ao namespace dos Hubs de Eventos. Para copiar a cadeia de conexão primária para seu namespace, siga estas etapas:

1. Na página Namespace dos Hubs de Eventos, selecione Políticas de acesso compartilhado no menu à esquerda.

2. Na página Políticas de acesso compartilhado, selecione RootManageSharedAccessKey.

3. Na janela Política: RootManageSharedAccessKey, selecione o botão Copiar próximo à Cadeia de Conexão Primária para copiar a cadeia de conexão na área de transferência para uso posterior. Cole esse valor no Bloco de notas ou em outro local temporário.

   

   Você pode usar essa página para copiar a chave primária, a chave secundária, a cadeia de conexão primária e a cadeia de conexão secundária.

Adicione uma classe chamada Sender e adicione o seguinte código à classe:

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

Adicione uma classe chamada Sender e adicione o seguinte código à classe:

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

Adicionar o código para publicar eventos no hub de eventos

Adicione um método chamado publishEvents à classe Sender:

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

Ao desenvolver localmente, verifique se a conta de usuário que está acessando os dados de blob tem as permissões corretas. Você precisará do Colaborador de Dados do Blob de Armazenamento para ler e gravar os dados de blob. Para atribuir essa função a si mesmo, você precisará receber a atribuição da função Administrador de Acesso do Usuário ou de outra função que inclua a ação Microsoft.Authorization/roleAssignments/write. É possível atribuir funções RBAC do Azure a um usuário usando o portal do Azure, a CLI do Azure ou o Azure PowerShell. Você pode saber mais sobre os escopos disponíveis para atribuições de função na página de visão geral do escopo.

Nesse cenário, você atribuirá permissões à sua conta de usuário, no escopo da conta de armazenamento, para seguir o Princípio do Privilégio Mínimo. Essa prática fornece aos usuários apenas as permissões mínimas necessárias e cria ambientes de produção mais seguros.

O exemplo a seguir atribuirá a função de Colaborador de Dados do Blob de Armazenamento à sua conta de usuário, que fornece acesso de leitura e gravação aos dados de blob na sua conta de armazenamento.

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>

Obtenha a cadeia de conexão para a conta de armazenamento.

Anote a cadeia de conexão e o nome do contêiner. Você os usará no código de recebimento.

Adicione as dependências a seguir ao arquivo 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.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>

Adicione as dependências a seguir ao arquivo pom.xml.

• 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. Adicione as instruções import a seguir à parte superior do arquivo 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. Crie uma classe chamada Receiver e adicione as variáveis de cadeia de caracteres a seguir à classe. Substitua os espaços reservados pelos valores corretos.

   Importante

   Substitua os espaços reservados pelos valores corretos.

   • <NAMESPACE NAME> pelo nome do namespace dos Hubs de Eventos.
   • <EVENT HUB NAME> pelo nome do hub de eventos no namespace.

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

3. Adicione o método main a seguir à classe.

   Importante

   Substitua os espaços reservados pelos valores corretos.

   • <STORAGE ACCOUNT NAME> com o nome da sua conta de Armazenamento do Azure.
   • <CONTAINER NAME> com o nome do contêiner de blob na conta de armazenamento

   // 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. Adicione as instruções import a seguir à parte superior do arquivo 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. Crie uma classe chamada Receiver e adicione as variáveis de cadeia de caracteres a seguir à classe. Substitua os espaços reservados pelos valores corretos.

   Importante

   Substitua os espaços reservados pelos valores corretos.

   • <Event Hubs namespace connection string> pela cadeia de conexão do namespace dos Hubs de Eventos. Atualizar
   • <Event hub name> pelo nome do hub de eventos no namespace.
   • <Storage connection string> pela cadeia de conexão da conta de armazenamento do Azure.
   • <Storage container name> pelo nome do contêiner no Armazenamento de Blobs do Azure.

   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. Adicione o método main a seguir à classe.

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