ServiceBusClientBuilder Klasse

Diese Klasse stellt eine Fluent Builder-API bereit, um die Instanziierung von Clients zum Senden und Empfangen von Nachrichten an/von Service Bus-Entitäten zu unterstützen.

Anmeldeinformationen sind erforderlich, um Vorgänge für Azure Service Bus auszuführen. Sie können mit einer der folgenden Methoden festgelegt werden:

• connectionString(String connectionString)mit einem Verbindungszeichenfolge zum Service Bus-Namespace.
• credential(String fullyQualifiedNamespace, TokenCredential credential), und credential(String fullyQualifiedNamespace, AzureSasCredential credential)credential(String fullyQualifiedNamespace, AzureNamedKeyCredential credential) Überladungen können mit den entsprechenden Anmeldeinformationen verwendet werden, die Zugriff auf den vollqualifizierten Service Bus-Namespace haben.
• credential(TokenCredential credential), credential(AzureSasCredential credential)- und credential(AzureNamedKeyCredential credential) -Überladungen können mit den jeweiligen Anmeldeinformationen verwendet werden. fullyQualifiedNamespace(String fullyQualifiedNamespace)muss festgelegt werden .

Die in den folgenden Beispielen verwendeten Anmeldeinformationen dienen DefaultAzureCredential der Authentifizierung. Es eignet sich für die meisten Szenarien, einschließlich lokaler Entwicklungs- und Produktionsumgebungen. Darüber hinaus wird empfohlen, die verwaltete Identität für die Authentifizierung in Produktionsumgebungen zu verwenden. Weitere Informationen zu verschiedenen Authentifizierungsmethoden und den entsprechenden Anmeldeinformationstypen finden Sie in der Azure Identity-Dokumentation.

Clients und Unterbuilder

ServiceBusClientBuilder kann mehrere Clients instanziieren. Der zu instanziierende Client hängt davon ab, ob Benutzer Nachrichten veröffentlichen oder empfangen und ob die Entität Service Bus-Sitzungen aktiviert hat.

• Senden von Nachrichten: Verwenden Sie den sender() Unter-Generator, um und ServiceBusSenderClientzu erstellenServiceBusSenderAsyncClient.
• Empfangen von Nachrichten: Verwenden Sie den receiver() Unter-Generator, um und ServiceBusReceiverAsyncClientzu erstellenServiceBusReceiverAsyncClient.
• Empfangen von Nachrichten von einer sitzungsfähigen Service Bus-Entität: Verwenden Sie den sessionReceiver() Unter-Generator, um und ServiceBusSessionReceiverClientzu erstellenServiceBusSessionReceiverAsyncClient.
• Empfangen von Nachrichten mithilfe eines rückrufbasierten Prozessors: Verwenden Sie den processor() Unter-Generator, um zu erstellen ServiceBusProcessorClient.
• Empfangen von Nachrichten von einer sitzungsfähigen Service Bus-Entität mithilfe eines rückrufbasierten Prozessors : Verwenden Sie den sessionProcessor() Unter-Generator, um zu erstellen ServiceBusProcessorClient.

Senden von Nachrichten

Beispiel: Instanziieren eines synchronen Absenders und Senden einer Nachricht

Im folgenden Codebeispiel wird die Erstellung des synchronen Clients ServiceBusSenderClient und das Senden einer Nachricht veranschaulicht. Der fullyQualifiedNamespace ist der Hostname des Service Bus-Namespaces. Sie wird im Bereich "Essentials" aufgeführt, nachdem Sie über das Azure-Portal zum Service Bus-Namespace navigieren. Die verwendeten Anmeldeinformationen basieren DefaultAzureCredential darauf, dass sie häufig verwendete Anmeldeinformationen in Bereitstellung und Entwicklung kombiniert und die zu verwendenden Anmeldeinformationen basierend auf der ausgeführten Umgebung auswählen. Wenn die Leistung wichtig ist, sollten Sie verwenden ServiceBusMessageBatch , um mehrere Nachrichten gleichzeitig zu veröffentlichen.

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

Nachrichtenverbrauch

Es gibt mehrere Clients für die Verwendung von Nachrichten von einer Service Bus-Entität (für die Service Bus-Sitzungen nicht aktiviert sind).

Beispiel: Instanziieren eines asynchronen Empfängers

Im folgenden Codebeispiel wird das Erstellen eines asynchronen Empfängers veranschaulicht. Die verwendeten Anmeldeinformationen dienen DefaultAzureCredential der Authentifizierung. Es eignet sich für die meisten Szenarien, einschließlich lokaler Entwicklungs- und Produktionsumgebungen. PEEK_LOCK und disableAutoComplete() werden dringend empfohlen, damit Benutzer die Kontrolle über die Nachrichtenabwicklung haben.

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

Beispiel: Instanziieren ServiceBusProcessorClient

Im folgenden Codebeispiel wird das Erstellen eines Prozessorclients veranschaulicht. Der Prozessorclient wird für die meisten Produktionsszenarien empfohlen, da er die Verbindungswiederherstellung bietet. Die verwendeten Anmeldeinformationen dienen DefaultAzureCredential der Authentifizierung. Es eignet sich für die meisten Szenarien, einschließlich lokaler Entwicklungs- und Produktionsumgebungen. PEEK_LOCK und disableAutoComplete() werden dringend empfohlen, damit Benutzer die Kontrolle über die Nachrichtenabwicklung haben.

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

Verwenden von Nachrichten von einer sitzungsfähigen Service Bus-Entität

Service Bus unterstützt die gemeinsame und geordnete Verarbeitung ungebundener Nachrichtensequenzen über Service Bus-Sitzungen. Sitzungen können als FIFO-Verarbeitung (First In, First Out) von Nachrichten verwendet werden. Warteschlangen und Themen/Abonnements unterstützen Service Bus-Sitzungen, müssen jedoch zum Zeitpunkt der Entitätserstellung aktiviert werden.

Beispiel: Senden einer Nachricht an eine sitzungsfähige Warteschlange

Der folgende Codeausschnitt veranschaulicht das Senden einer Nachricht an eine Service Bus-Sitzungswarteschlange , die aktiviert ist. Durch Festlegen setMessageId(String messageId) der Eigenschaft auf "Greetings" wird die Nachricht an eine Service Bus-Sitzung mit der ID "Greetings" gesendet.

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

Beispiel: Empfangen von Nachrichten aus der ersten verfügbaren Sitzung

Um Nachrichten aus der ersten verfügbaren Sitzung zu verarbeiten, wechseln Sie zum ServiceBusSessionReceiverClientBuilder Sitzungsempfängerclient, und erstellen Sie diesen. Verwenden Sie acceptNextSession() , um die erste verfügbare Sitzung zum Verarbeiten von Nachrichten zu finden.

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

Beispiel: Verarbeiten von Nachrichten aus allen Sitzungen

Im folgenden Codebeispiel wird die Erstellung veranschaulicht, die ServiceBusProcessorClient alle verfügbaren Sitzungen in der Warteschlange verarbeitet. ServiceBusSessionProcessorClientBuilder#maxConcurrentSessions(int) gibt an, wie viele Sitzungen der Prozessor gleichzeitig verarbeitet. Die verwendeten Anmeldeinformationen dienen DefaultAzureCredential der Authentifizierung. Es eignet sich für die meisten Szenarien, einschließlich lokaler Entwicklungs- und Produktionsumgebungen. PEEK_LOCK und disableAutoComplete() werden dringend empfohlen, damit Benutzer die Kontrolle über die Nachrichtenabwicklung haben.

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

Verbindungsfreigabe

Für das Erstellen einer Verbindung mit Service Bus sind Ressourcen erforderlich. Wenn Ihre Architektur es zulässt, sollte eine Anwendung eine Verbindung zwischen Clients freigeben, die durch die Gemeinsame Nutzung des Generators der obersten Ebene erreicht werden kann, wie unten gezeigt.

Freigeben einer Verbindung zwischen Clients

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