ServiceBusClientBuilder Classe

Essa classe fornece uma API de construtor fluente para auxiliar na instanciação de clientes para enviar e receber mensagens de/para entidades do Barramento de Serviço.

As credenciais são necessárias para executar operações em Barramento de Serviço do Azure. Eles podem ser definidos usando um dos seguintes métodos:

• connectionString(String connectionString)com um cadeia de conexão para o namespace do Barramento de Serviço.
• credential(String fullyQualifiedNamespace, TokenCredential credential)As sobrecargas , credential(String fullyQualifiedNamespace, AzureSasCredential credential)e credential(String fullyQualifiedNamespace, AzureNamedKeyCredential credential) podem ser usadas com as respectivas credenciais que têm acesso ao namespace do Barramento de Serviço totalmente qualificado.
• credential(TokenCredential credential)As sobrecargas , credential(AzureSasCredential credential)e credential(AzureNamedKeyCredential credential) podem ser usadas com suas respectivas credenciais. fullyQualifiedNamespace(String fullyQualifiedNamespace)deve ser definido.

A credencial usada nos exemplos a seguir é DefaultAzureCredential para autenticação. É apropriado para a maioria dos cenários, incluindo ambientes locais de desenvolvimento e produção. Além disso, é recomendável usar a identidade gerenciada para autenticação em ambientes de produção. Você pode encontrar mais informações sobre diferentes maneiras de autenticação e seus tipos de credenciais correspondentes na documentação da Identidade do Azure".

Clientes e subconstrutores

ServiceBusClientBuilder pode criar uma instância de vários clientes. O cliente a ser instanciado depende se os usuários estão publicando ou recebendo mensagens e se a entidade tem sessões do Barramento de Serviço habilitadas.

• Enviando mensagens: use o sender() subconstrutor para criar ServiceBusSenderAsyncClient e ServiceBusSenderClient.
• Recebendo mensagens: use o receiver() subconstrutor para criar ServiceBusReceiverAsyncClient e ServiceBusReceiverAsyncClient.
• Recebendo mensagens de uma entidade do Barramento de Serviço habilitada para sessão: use o sessionReceiver() subconstrutor para criar ServiceBusSessionReceiverAsyncClient e ServiceBusSessionReceiverClient.
• Recebendo mensagens usando um processador baseado em retorno de chamada: use o processor() subconstrutor para criar ServiceBusProcessorClient.
• Recebendo mensagens de uma entidade do Barramento de Serviço habilitada para sessão usando um processador baseado em retorno de chamada : use o sessionProcessor() subconstrutor para criar ServiceBusProcessorClient.

Envio de mensagens

Exemplo: instanciar um remetente síncrono e enviar uma mensagem

O exemplo de código a seguir demonstra a criação do cliente ServiceBusSenderClient síncrono e o envio de uma mensagem. O fullyQualifiedNamespace é o nome do host do namespace do Barramento de Serviço. Ele é listado no painel "Essentials" depois de navegar até o namespace do Barramento de Serviço por meio do Portal do Azure. A credencial usada é DefaultAzureCredential porque combina credenciais comumente usadas na implantação e desenvolvimento e escolhe a credencial a ser usada com base em seu ambiente em execução. Quando o desempenho for importante, considere usar ServiceBusMessageBatch para publicar várias mensagens ao mesmo tempo.

TokenCredential credential = new DefaultAzureCredentialBuilder().build();

 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 ServiceBusSenderClient sender = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, credential)
     .sender()
     .queueName(queueName)
     .buildClient();

 sender.sendMessage(new ServiceBusMessage("Foo bar"));

Consumindo mensagens

Há vários clientes para consumir mensagens de uma entidade do Barramento de Serviço (que não tem sessões do Barramento de Serviço habilitadas).

Exemplo: instanciar um receptor assíncrono

O exemplo de código abaixo demonstra a criação de um receptor assíncrono. A credencial usada é DefaultAzureCredential para autenticação. É apropriado para a maioria dos cenários, incluindo ambientes locais de desenvolvimento e produção. PEEK_LOCK e disableAutoComplete() são altamente recomendados para que os usuários tenham controle sobre a liquidação de mensagens.

TokenCredential credential = new DefaultAzureCredentialBuilder().build();

 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 // 'disableAutoComplete' indicates that users will explicitly settle their message.
 ServiceBusReceiverAsyncClient asyncReceiver = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, credential)
     .receiver()
     .disableAutoComplete()
     .queueName(queueName)
     .buildAsyncClient();

 // When users are done with the receiver, dispose of the receiver.
 // Clients should be long-lived objects as they require resources
 // and time to establish a connection to the service.
 asyncReceiver.close();

Exemplo: Instanciar ServiceBusProcessorClient

O exemplo de código abaixo demonstra a criação de um cliente de processador. O cliente do processador é recomendado para a maioria dos cenários de produção porque oferece recuperação de conexão. A credencial usada é DefaultAzureCredential para autenticação. É apropriado para a maioria dos cenários, incluindo ambientes locais de desenvolvimento e produção. PEEK_LOCK e disableAutoComplete() são altamente recomendados para que os usuários tenham controle sobre a liquidação de mensagens.

// Function that gets called whenever a message is received.
 Consumer<ServiceBusReceivedMessageContext> processMessage = context -> {
     final ServiceBusReceivedMessage message = context.getMessage();
     // Randomly complete or abandon each message. Ideally, in real-world scenarios, if the business logic
     // handling message reaches desired state such that it doesn't require Service Bus to redeliver
     // the same message, then context.complete() should be called otherwise context.abandon().
     final boolean success = Math.random() < 0.5;
     if (success) {
         try {
             context.complete();
         } catch (RuntimeException error) {
             System.out.printf("Completion of the message %s failed.%n Error: %s%n",
                 message.getMessageId(), error);
         }
     } else {
         try {
             context.abandon();
         } catch (RuntimeException error) {
             System.out.printf("Abandoning of the message %s failed.%nError: %s%n",
                 message.getMessageId(), error);
         }
     }
 };

 // Sample code that gets called if there's an error
 Consumer<ServiceBusErrorContext> processError = errorContext -> {
     if (errorContext.getException() instanceof ServiceBusException) {
         ServiceBusException exception = (ServiceBusException) errorContext.getException();

         System.out.printf("Error source: %s, reason %s%n", errorContext.getErrorSource(),
             exception.getReason());
     } else {
         System.out.printf("Error occurred: %s%n", errorContext.getException());
     }
 };

 TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();

 // Create the processor client via the builder and its sub-builder
 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, tokenCredential)
     .processor()
     .queueName(queueName)
     .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
     .disableAutoComplete()  // Make sure to explicitly opt in to manual settlement (e.g. complete, abandon).
     .processMessage(processMessage)
     .processError(processError)
     .disableAutoComplete()
     .buildProcessorClient();

 // Starts the processor in the background. Control returns immediately.
 processorClient.start();

 // Stop processor and dispose when done processing messages.
 processorClient.stop();
 processorClient.close();

Consumindo mensagens de uma entidade do Barramento de Serviço habilitada para sessão

O Barramento de Serviço dá suporte ao tratamento conjunto e ordenado de sequências não associadas de mensagens por meio de sessões do Barramento de Serviço. As sessões podem ser usadas como um processamento FIFO (primeiro a entrar, primeiro a sair) de mensagens. Filas e tópicos/assinaturas dão suporte a sessões do Barramento de Serviço, no entanto, ele deve ser habilitado no momento da criação da entidade.

Exemplo: enviando uma mensagem para uma fila habilitada para sessão

O snippet a seguir demonstra o envio de uma mensagem para uma fila habilitada para sessões do Barramento de Serviço . Definir setMessageId(String messageId) a propriedade como "saudações" enviará a mensagem para uma sessão do Barramento de Serviço com uma ID de "saudações".

// 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 ServiceBusSenderClient sender = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, new DefaultAzureCredentialBuilder().build())
     .sender()
     .queueName(sessionEnabledQueueName)
     .buildClient();

 // Setting sessionId publishes that message to a specific session, in this case, "greeting".
 ServiceBusMessage message = new ServiceBusMessage("Hello world")
     .setSessionId("greetings");

 sender.sendMessage(message);

 // Dispose of the sender.
 sender.close();

Exemplo: receber mensagens da primeira sessão disponível

Para processar mensagens da primeira sessão disponível, alterne para ServiceBusSessionReceiverClientBuilder e compile o cliente receptor da sessão. Use acceptNextSession() para localizar a primeira sessão disponível para processar mensagens.

TokenCredential credential = new DefaultAzureCredentialBuilder().build();

 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 // 'disableAutoComplete' indicates that users will explicitly settle their message.
 ServiceBusSessionReceiverAsyncClient sessionReceiver = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, credential)
     .sessionReceiver()
     .disableAutoComplete()
     .queueName(sessionEnabledQueueName)
     .buildAsyncClient();

 // Creates a client to receive messages from the first available session. It waits until
 // AmqpRetryOptions.getTryTimeout() elapses. If no session is available within that operation timeout, it
 // completes with a retriable error. Otherwise, a receiver is returned when a lock on the session is acquired.
 Mono<ServiceBusReceiverAsyncClient> receiverMono = sessionReceiver.acceptNextSession();

 Flux<Void> receiveMessagesFlux = Flux.usingWhen(receiverMono,
     receiver -> receiver.receiveMessages().flatMap(message -> {
         System.out.println("Received message: " + message.getBody());

         // Explicitly settle the message via complete, abandon, defer, dead-letter, etc.
         if (isMessageProcessed) {
             return receiver.complete(message);
         } else {
             return receiver.abandon(message);
         }
     }),
     receiver -> Mono.fromRunnable(() -> {
         // Dispose of the receiver and sessionReceiver when done receiving messages.
         receiver.close();
         sessionReceiver.close();
     }));

 // This is a non-blocking call that moves onto the next line of code after setting up and starting the receive
 // operation. Customers can keep a reference to `subscription` and dispose of it when they want to stop
 // receiving messages.
 Disposable subscription = receiveMessagesFlux.subscribe(unused -> {
 }, error -> System.out.println("Error occurred: " + error),
     () -> System.out.println("Receiving complete."));

Exemplo: processar mensagens de todas as sessões

O exemplo de código a seguir demonstra a criação do ServiceBusProcessorClient que processa todas as sessões disponíveis na fila. ServiceBusSessionProcessorClientBuilder#maxConcurrentSessions(int) indica quantas sessões o processador processará ao mesmo tempo. A credencial usada é DefaultAzureCredential para autenticação. É apropriado para a maioria dos cenários, incluindo ambientes locais de desenvolvimento e produção. PEEK_LOCK e disableAutoComplete() são altamente recomendados para que os usuários tenham controle sobre a liquidação de mensagens.

// Function that gets called whenever a message is received.
 Consumer<ServiceBusReceivedMessageContext> onMessage = context -> {
     ServiceBusReceivedMessage message = context.getMessage();
     System.out.printf("Processing message. Session: %s, Sequence #: %s. Contents: %s%n",
         message.getSessionId(), message.getSequenceNumber(), message.getBody());
 };

 Consumer<ServiceBusErrorContext> onError = context -> {
     System.out.printf("Error when receiving messages from namespace: '%s'. Entity: '%s'%n",
         context.getFullyQualifiedNamespace(), context.getEntityPath());

     if (context.getException() instanceof ServiceBusException) {
         ServiceBusException exception = (ServiceBusException) context.getException();

         System.out.printf("Error source: %s, reason %s%n", context.getErrorSource(),
             exception.getReason());
     } else {
         System.out.printf("Error occurred: %s%n", context.getException());
     }
 };

 TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();

 // Create the processor client via the builder and its sub-builder
 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 ServiceBusProcessorClient sessionProcessor = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, tokenCredential)
     .sessionProcessor()
     .queueName(sessionEnabledQueueName)
     .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
     .disableAutoComplete()
     .maxConcurrentSessions(2)
     .processMessage(onMessage)
     .processError(onError)
     .buildProcessorClient();

 // Starts the processor in the background. Control returns immediately.
 sessionProcessor.start();

 // Stop processor and dispose when done processing messages.
 sessionProcessor.stop();
 sessionProcessor.close();

Compartilhamento de conexão

A criação de uma conexão com o Barramento de Serviço requer recursos. Se sua arquitetura permitir, um aplicativo deverá compartilhar a conexão entre clientes que pode ser obtida compartilhando o construtor de nível superior, conforme mostrado abaixo.

Compartilhando uma conexão entre clientes

TokenCredential credential = new DefaultAzureCredentialBuilder().build();

 // 'fullyQualifiedNamespace' will look similar to "{your-namespace}.servicebus.windows.net"
 // Any clients created from this builder will share the underlying connection.
 ServiceBusClientBuilder sharedConnectionBuilder = new ServiceBusClientBuilder()
     .credential(fullyQualifiedNamespace, credential);

 // Create receiver and sender which will share the connection.
 ServiceBusReceiverClient receiver = sharedConnectionBuilder
     .receiver()
     .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
     .queueName(queueName)
     .buildClient();
 ServiceBusSenderClient sender = sharedConnectionBuilder
     .sender()
     .queueName(queueName)
     .buildClient();

 // Use the clients and finally close them.
 try {
     sender.sendMessage(new ServiceBusMessage("payload"));
     receiver.receiveMessages(1);
 } finally {
     // Clients should be long-lived objects as they require resources
     // and time to establish a connection to the service.
     sender.close();
     receiver.close();
 }