See: Description
Package | Description |
---|---|
com.azure.messaging.servicebus |
Package containing classes for creating
ServiceBusSenderAsyncClient ,
ServiceBusProcessorClient , and
ServiceBusReceiverAsyncClient to perform operations on Azure Service Bus. |
com.azure.messaging.servicebus.administration |
Package containing classes managing resources in Azure Service Bus.
|
com.azure.messaging.servicebus.administration.models |
Package containing classes models for
ServiceBusAdministrationClient
and ServiceBusAdministrationAsyncClient used for managing
Queues, Topics, and Subscriptions. |
com.azure.messaging.servicebus.models |
Model classes for interacting with Azure Service Bus.
|
Microsoft Azure Service Bus is a fully managed enterprise integration message broker. Service Bus can decouple applications and services. Service Bus offers a reliable and secure platform for asynchronous transfer of data and state. Data is transferred between different applications and services using messages. If you would like to know more about Azure Service Bus, you may wish to review: What is Service Bus
The Azure Service Bus client library allows for sending and receiving of Azure Service Bus messages and may be used to: - Transfer business data, such as sales or purchase orders, journals, or inventory movements. - Decouple applications to improve reliability and scalability of applications and services. Clients and services don't have to be online at the same time. - Enable 1:n relationships between publishers and subscribers. - Implement workflows that require message ordering or message deferral.
Source code | API reference documentation | Product documentation| Samples
To quickly create the needed Service Bus resources in Azure and to receive a connection string for them, you can deploy our sample template by clicking:
<dependency>
<groupId>com.azure</groupId>
<artifactId>azure-messaging-servicebus</artifactId>
<version>7.2.0-beta.1</version>
</dependency>
For the Service Bus client library to interact with Service Bus, it will need to understand how to connect and authorize with it.
The easiest means for authenticating is to use a connection string, which automatically created when creating a Service Bus namespace. If you aren't familiar with shared access policies in Azure, you may wish to follow the step-by-step guide to get a Service Bus connection string.
Both the asynchronous and synchronous Service Bus sender and receiver clients are instantiated using
ServiceBusClientBuilder
. The snippets below create a synchronous Service Bus sender and an asynchronous receiver,
respectively.
ServiceBusSenderClient sender = new ServiceBusClientBuilder()
.connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
.sender()
.queueName("<< QUEUE NAME >>")
.buildClient();
ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder()
.connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
.receiver()
.topicName("<< TOPIC NAME >>")
.subscriptionName("<< SUBSCRIPTION NAME >>")
.buildAsyncClient();
Azure SDK for Java supports the Azure Identity package, making it simple to get credentials from the Microsoft identity platform. First, add the package:
<dependency>
<groupId>com.azure</groupId>
<artifactId>azure-identity</artifactId>
<version>1.2.5</version>
</dependency>
azure-messaging-servicebus
before azure-identity
client libraries. This
issue is resolved with azure-identity:1.2.1
.
Check here for more details.The implemented ways to request a credential are under the com.azure.identity.credential
package. The sample below
shows how to use an Azure Active Directory (AAD) application client secret to authorize with Azure Service Bus.
Authorization is easiest using DefaultAzureCredential. It finds the best credential to use in its running environment. For more information about using Azure Active Directory authorization with Service Bus, please refer to the associated documentation.
Use the returned token credential to authenticate the client:
TokenCredential credential = new DefaultAzureCredentialBuilder()
.build();
ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder()
.credential("<<fully-qualified-namespace>>", credential)
.receiver()
.queueName("<<queue-name>>")
.buildAsyncClient();
You can interact with the primary resource types within a Service Bus Namespace, of which multiple can exist and on which actual message transmission takes place. The namespace often serves as an application container:
The builder ServiceBusClientBuilder
is used to create all the Service Bus clients.
ServiceBusSenderClient
A synchronous sender responsible for sending
ServiceBusMessage
to specific queue or topic on Azure Service Bus.ServiceBusSenderAsyncClient
A asynchronous sender responsible for sending
ServiceBusMessage
to specific queue or topic on Azure Service Bus.ServiceBusReceiverClient
A synchronous receiver responsible for receiving
ServiceBusMessage
from a specific queue or topic on Azure Service Bus.ServiceBusReceiverAsyncClient
A asynchronous receiver responsible for
receiving ServiceBusMessage
from a specific queue or topic on Azure Service Bus.You'll need to create an asynchronous ServiceBusSenderAsyncClient
or a synchronous
ServiceBusSenderClient
to send messages. Each sender can send messages to either a queue or
a topic.
The snippet below creates a synchronous ServiceBusSenderClient
to publish a message to a
queue.
ServiceBusSenderClient sender = new ServiceBusClientBuilder()
.connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
.sender()
.queueName("<< QUEUE NAME >>")
.buildClient();
List<ServiceBusMessage> messages = Arrays.asList(
new ServiceBusMessage("Hello world").setMessageId("1"),
new ServiceBusMessage("Bonjour").setMessageId("2"));
sender.sendMessages(messages);
// When you are done using the sender, dispose of it.
sender.close();
To receive messages, you will need to create a ServiceBusProcessorClient
with callbacks for incoming messages and any error that occurs in the process. You can then start and stop the client as required.
By default, the autoComplete
feature is enabled on the processor client which means that after executing your callback for the message, the client will complete the message i.e. remove it from the queue/subscription. If your callback throws an error, then the client will abandon the message i.e. make it available to be received again. You can disable this feature when creating the processor client.
// Sample code that processes a single message
Consumer<ServiceBusReceivedMessageContext> processMessage = messageContext -> {
try {
System.out.println(messageContext.getMessage().getMessageId());
// other message processing code
messageContext.complete();
} catch (Exception ex) {
messageContext.abandon();
}
};
// Sample code that gets called if there's an error
Consumer<ServiceBusErrorContext> processError = errorContext -> {
System.err.println("Error occurred while receiving message: " + errorContext.getException());
};
// create the processor client via the builder and its sub-builder
ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
.connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
.processor()
.queueName("<< QUEUE NAME >>")
.processMessage(processMessage)
.processError(processError)
.disableAutoComplete()
.buildProcessorClient();
// Starts the processor in the background and returns immediately
processorClient.start();
There are four ways of settling messages using the methods on the message context passed to your callback. - Complete - causes the message to be deleted from the queue or topic. - Abandon - releases the receiver's lock on the message allowing for the message to be received by other receivers. - Defer - defers the message from being received by normal means. In order to receive deferred messages, the sequence number of the message needs to be retained. - Dead-letter - moves the message to the dead-letter queue. This will prevent the message from being received again. In order to receive messages from the dead-letter queue, a receiver scoped to the dead-letter queue is needed.
Using sessions requires you to create a session enabled queue or subscription. You can read more about how to configure this in "Message sessions".
Azure Service Bus sessions enable joint and ordered handling of unbounded sequences of related messages. Sessions can be
used in first in, first out (FIFO) and request-response patterns. Any sender can create a session when submitting
messages into a topic or queue by setting the ServiceBusMessage.setSessionId(String)
property to some
application-defined identifier that is unique to the session.
Unlike non-session-enabled queues or subscriptions, only a single receiver can read from a session at any time. When a receiver fetches a session, Service Bus locks the session for that receiver, and it has exclusive access to messages in that session.
Create a ServiceBusSenderClient
for a session enabled queue or topic subscription. Setting
ServiceBusMessage.setSessionId(String)
on a ServiceBusMessage
will publish the message to that session. If the
session does not exist, it is created.
// Setting sessionId publishes that message to a specific session, in this case, "greeting".
ServiceBusMessage message = new ServiceBusMessage("Hello world")
.setSessionId("greetings");
sender.sendMessage(message);
Receiving messages from sessions is similar to receiving messages from a non session enabled queue or subscription. The difference is in the builder and the class you use.
In non-session case, you would use the sub builder processor()
. In case of sessions, you would use the sub builder sessionProcessor()
. Both sub builders will create an instance of ServiceBusProcessorClient
configured to work on a session or a non-session Service Bus entity. In the case of the session processor, you can pass the maximum number of sessions you want the processor to process concurrently as well.
Azure Service Bus queues and topic subscriptions provide a secondary sub-queue, called a dead-letter queue (DLQ). The dead-letter queue doesn't need to be explicitly created and can't be deleted or otherwise managed independent of the main entity. For session enabled or non-session queue or topic subscriptions, the dead-letter receiver can be created the same way as shown below. Learn more about dead-letter queue here.
ServiceBusReceiverClient receiver = new ServiceBusClientBuilder()
.connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
.receiver() // Use this for session or non-session enabled queue or topic/subscriptions
.topicName("<< TOPIC NAME >>")
.subscriptionName("<< SUBSCRIPTION NAME >>")
.subQueue(SubQueue.DEAD_LETTER_QUEUE)
.buildClient();
The creation of physical connection to Service Bus requires resources. An application should share the connection
between clients which can be achieved by sharing the top level builder as shown below.
ServiceBusClientBuilder sharedConnectionBuilder = new ServiceBusClientBuilder()
.connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>");
// Create receiver and sender which will share the connection.
ServiceBusReceiverClient receiver = sharedConnectionBuilder
.receiver()
.queueName("<< QUEUE NAME >>")
.buildClient();
ServiceBusSenderClient sender = sharedConnectionBuilder
.sender()
.queueName("<< QUEUE NAME >>")
.buildClient();
}
When to use 'ServiceBusProcessorClient', 'ServiceBusReceiverClient' or ServiceBusReceiverAsyncClient? The processor
is built using 'ServiceBusReceiverAsyncClient', it provides a convenient way of receiving messages with default
auto complete and auto renew of message locks in 'PEEK_LOCK' mode. The processor is appropriate where the
applications have not made complete move to async receiver client and want to process message in synchronous mode.
The processor receives messages forever because it recovers from the network errors internally.
'ServiceBusProcessorClient:processMessage()' function call is made for each message. Alternatively, You can also use
'ServiceBusReceiverClient', it is a lower level client and provides a wider range of APIs. If async processing is
suitable for your application, you can use 'ServiceBusReceiverAsyncClient'.
Azure SDK for Java offers a consistent logging story to help aid in troubleshooting application errors and expedite their resolution. The logs produced will capture the flow of an application before reaching the terminal state to help locate the root issue. View the logging wiki for guidance about enabling logging.
If enabling client logging is not enough to diagnose your issues. You can enable logging to a file in the underlying
AMQP library, Qpid Proton-J. Qpid Proton-J uses java.util.logging
. You can enable logging by
create a configuration file with the contents below. Or set proton.trace.level=ALL
and whichever configuration options
you want for the java.util.logging.Handler
implementation. Implementation classes and their options can be found in
Java 8 SDK javadoc.
To trace the AMQP transport frames, set the environment variable: PN_TRACE_FRM=1
.
The configuration file below logs trace output from proton-j to the file "proton-trace.log".
handlers=java.util.logging.FileHandler
.level=OFF
proton.trace.level=ALL
java.util.logging.FileHandler.level=ALL
java.util.logging.FileHandler.pattern=proton-trace.log
java.util.logging.FileHandler.formatter=java.util.logging.SimpleFormatter
java.util.logging.SimpleFormatter.format=[%1$tF %1$tr] %3$s %4$s: %5$s %n
This is a general exception for AMQP related failures, which includes the AMQP errors as ErrorCondition
and the
context that caused this exception as AmqpErrorContext
. isTransient
is a boolean indicating if the exception is a
transient error or not. If a transient AMQP exception occurs, the client library retries the operation as many times
as the AmqpRetryOptons allows. Afterwords, the operation fails and an exception is propagated back
to the user.
AmqpErrorCondition
contains error conditions common to the AMQP protocol and used by Azure
services. When an AMQP exception is thrown, examining the error condition field can inform developers as to why the AMQP
exception occurred and if possible, how to mitigate this exception. A list of all the AMQP exceptions can be found in
OASIS AMQP Version 1.0 Transport Errors.
The recommended way to solve the specific exception the AMQP exception represents is to follow the Service Bus Messaging Exceptions guidance.
Beyond those discussed, the Azure Service Bus client library offers support for many additional scenarios to help take advantage of the full feature set of the Azure Service Bus service. In order to help explore some of these scenarios, the following set of sample is available here.
If you would like to become an active contributor to this project please refer to our Contribution Guidelines for more information.
Copyright © 2021 Microsoft Corporation. All rights reserved.