Skicka meddelanden till och ta emot meddelanden från Azure Service Bus köer (Java)

Artikel
11/30/2021
6 minuter för att läsa

I den här snabbstarten skapar du en Java-app för att skicka meddelanden till och ta emot meddelanden från en Azure Service Bus-kö.

Anteckning

Den här snabbstarten innehåller stegvisa instruktioner för ett enkelt scenario där meddelanden skickas till en Service Bus och tar emot dem. Du hittar förbyggda Java-exempel för Azure Service Bus på lagringsplatsen Azure SDK för Java på GitHub.

Förutsättningar

En Azure-prenumeration. Du behöver ett Azure-konto för att genomföra kursen. Du kan aktivera dina MSDN-prenumerantförmåner eller registrera dig för ett kostnadsfritt konto.
Om du inte har någon kö att arbeta med följer du stegen i artikeln Använda Azure Portal för att skapa en Service Bus för att skapa en kö. Anteckna anslutningssträngen för Service Bus namnområdet och namnet på den kö som du skapade.
Installera Azure SDK för Java. Om du använder Eclipse kan du installera den Azure Toolkit for Eclipse som innehåller Azure SDK för Java. Du kan sedan lägga till Microsoft Azure-bibliotek för Java i projektet. Om du använder IntelliJ kan du se Installera Azure Toolkit for IntelliJ.

Skicka meddelanden till en kö

I det här avsnittet skapar du ett Java-konsolprojekt och lägger till kod för att skicka meddelanden till kön som du skapade tidigare.

Skapa ett Java-konsolprojekt

Skapa ett Java-projekt med Eclipse eller ett val av verktyg.

Konfigurera ditt program att använda Service Bus

Lägg till referenser till Azure Core och Azure Service Bus bibliotek.

Om du använder Eclipse och har skapat ett Java-konsolprogram konverterar du Ditt Java-projekt till maven: högerklicka på projektet i fönstret Package Explorer och välj Konfigurera Konvertera -> till Maven-projekt. Lägg sedan till beroenden till dessa två bibliotek enligt följande exempel.

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>org.myorg.sbusquickstarts</groupId>
    <artifactId>sbustopicqs</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <build>
        <sourceDirectory>src</sourceDirectory>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <release>15</release>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <dependencies>
        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-core</artifactId>
            <version>1.13.0</version>
        </dependency>
        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-messaging-servicebus</artifactId>
            <version>7.0.2</version>
        </dependency>
    </dependencies>
</project>

Lägga till kod för att skicka meddelanden till kön

Lägg till följande import -instruktioner i avsnittet i Java-filen.

import com.azure.messaging.servicebus.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.Arrays;
import java.util.List;

I klassen definierar du variabler som ska innehålla anslutningssträngen och könamnet enligt nedan:
```
static String connectionString = "<NAMESPACE CONNECTION STRING>";
static String queueName = "<QUEUE NAME>";    
```
Ersätt <NAMESPACE CONNECTION STRING> med anslutningssträngen till Service Bus namnområdet. Och ersätt <QUEUE NAME> med namnet på kön.

Lägg till en metod med sendMessage namnet i klassen för att skicka ett meddelande till kön.

static void sendMessage()
{
    // create a Service Bus Sender client for the queue 
    ServiceBusSenderClient senderClient = new ServiceBusClientBuilder()
            .connectionString(connectionString)
            .sender()
            .queueName(queueName)
            .buildClient();

    // send one message to the queue
    senderClient.sendMessage(new ServiceBusMessage("Hello, World!"));
    System.out.println("Sent a single message to the queue: " + queueName);        
}

Lägg till en metod med createMessages namnet i klassen för att skapa en lista med meddelanden. Vanligtvis får du dessa meddelanden från olika delar av ditt program. Här skapar vi en lista över exempelmeddelanden.

static List<ServiceBusMessage> createMessages()
{
    // create a list of messages and return it to the caller
    ServiceBusMessage[] messages = {
            new ServiceBusMessage("First message"),
            new ServiceBusMessage("Second message"),
            new ServiceBusMessage("Third message")
    };
    return Arrays.asList(messages);
}

Lägg till en metod med sendMessageBatch namnet method för att skicka meddelanden till kön som du skapade. Den här metoden skapar en för kön, anropar metoden för att hämta listan över meddelanden, förbereder en eller flera batchar och skickar ServiceBusSenderClient createMessages batcharna till kön.

static void sendMessageBatch()
{
    // create a Service Bus Sender client for the queue 
    ServiceBusSenderClient senderClient = new ServiceBusClientBuilder()
            .connectionString(connectionString)
            .sender()
            .queueName(queueName)
            .buildClient();

    // Creates an ServiceBusMessageBatch where the ServiceBus.
    ServiceBusMessageBatch messageBatch = senderClient.createMessageBatch();        

    // create a list of messages
    List<ServiceBusMessage> listOfMessages = createMessages();

    // We try to add as many messages as a batch can fit based on the maximum size and send to Service Bus when
    // the batch can hold no more messages. Create a new batch for next set of messages and repeat until all
    // messages are sent.        
    for (ServiceBusMessage message : listOfMessages) {
        if (messageBatch.tryAddMessage(message)) {
            continue;
        }

        // The batch is full, so we create a new batch and send the batch.
        senderClient.sendMessages(messageBatch);
        System.out.println("Sent a batch of messages to the queue: " + queueName);

        // create a new batch
        messageBatch = senderClient.createMessageBatch();

        // Add that message that we couldn't before.
        if (!messageBatch.tryAddMessage(message)) {
            System.err.printf("Message is too large for an empty batch. Skipping. Max size: %s.", messageBatch.getMaxSizeInBytes());
        }
    }

    if (messageBatch.getCount() > 0) {
        senderClient.sendMessages(messageBatch);
        System.out.println("Sent a batch of messages to the queue: " + queueName);
    }

    //close the client
    senderClient.close();
}

Ta emot meddelanden från en kö

I det här avsnittet lägger du till kod för att hämta meddelanden från kön.

Lägg till en metod med receiveMessages namnet för att ta emot meddelanden från kön. Den här metoden skapar ServiceBusProcessorClient en för kön genom att ange en hanterare för bearbetning av meddelanden och en annan för att hantera fel. Sedan startar den processorn, väntar i några sekunder, skriver ut de meddelanden som tas emot och stoppar och stänger sedan processorn.

Viktigt

Ersätt QueueTest QueueTest::processMessage i i koden med namnet på din klass.

// handles received messages
static void receiveMessages() throws InterruptedException
{
    CountDownLatch countdownLatch = new CountDownLatch(1);

    // Create an instance of the processor through the ServiceBusClientBuilder
    ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
        .connectionString(connectionString)
        .processor()
        .queueName(queueName)
        .processMessage(QueueTest::processMessage)
        .processError(context -> processError(context, countdownLatch))
        .buildProcessorClient();

    System.out.println("Starting the processor");
    processorClient.start();

    TimeUnit.SECONDS.sleep(10);
    System.out.println("Stopping and closing the processor");
    processorClient.close();        
}

Lägg till processMessage metoden för att bearbeta ett meddelande som tas emot från Service Bus prenumerationen.

private static void processMessage(ServiceBusReceivedMessageContext context) {
    ServiceBusReceivedMessage message = context.getMessage();
    System.out.printf("Processing message. Session: %s, Sequence #: %s. Contents: %s%n", message.getMessageId(),
        message.getSequenceNumber(), message.getBody());
}

Lägg till processError metoden för att hantera felmeddelanden.

private static void processError(ServiceBusErrorContext context, CountDownLatch countdownLatch) {
    System.out.printf("Error when receiving messages from namespace: '%s'. Entity: '%s'%n",
        context.getFullyQualifiedNamespace(), context.getEntityPath());

    if (!(context.getException() instanceof ServiceBusException)) {
        System.out.printf("Non-ServiceBusException occurred: %s%n", context.getException());
        return;
    }

    ServiceBusException exception = (ServiceBusException) context.getException();
    ServiceBusFailureReason reason = exception.getReason();

    if (reason == ServiceBusFailureReason.MESSAGING_ENTITY_DISABLED
        || reason == ServiceBusFailureReason.MESSAGING_ENTITY_NOT_FOUND
        || reason == ServiceBusFailureReason.UNAUTHORIZED) {
        System.out.printf("An unrecoverable error occurred. Stopping processing with reason %s: %s%n",
            reason, exception.getMessage());

        countdownLatch.countDown();
    } else if (reason == ServiceBusFailureReason.MESSAGE_LOCK_LOST) {
        System.out.printf("Message lock lost for message: %s%n", context.getException());
    } else if (reason == ServiceBusFailureReason.SERVICE_BUSY) {
        try {
            // Choosing an arbitrary amount of time to wait until trying again.
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            System.err.println("Unable to sleep for period of time");
        }
    } else {
        System.out.printf("Error source %s, reason %s, message: %s%n", context.getErrorSource(),
            reason, context.getException());
    }
}

Uppdatera metoden main för att anropa metoderna , och och för att skapa sendMessage sendMessageBatch receiveMessages InterruptedException .

public static void main(String[] args) throws InterruptedException {        
    sendMessage();
    sendMessageBatch();
    receiveMessages();
}

Kör appen

När du kör programmet visas följande meddelanden i konsolfönstret.

Sent a single message to the queue: myqueue
Sent a batch of messages to the queue: myqueue
Starting the processor
Processing message. Session: 88d961dd801f449e9c3e0f8a5393a527, Sequence #: 1. Contents: Hello, World!
Processing message. Session: e90c8d9039ce403bbe1d0ec7038033a0, Sequence #: 2. Contents: First message
Processing message. Session: 311a216a560c47d184f9831984e6ac1d, Sequence #: 3. Contents: Second message
Processing message. Session: f9a871be07414baf9505f2c3d466c4ab, Sequence #: 4. Contents: Third message
Stopping and closing the processor

På sidan Översikt för Service Bus namnrymden i Azure Portal kan du se antal inkommande och utgående meddelanden. Du kan behöva vänta i någon minut och sedan uppdatera sidan för att se de senaste värdena.

Välj kön på den här översiktssidan för att gå Service Bus sidan Kö. Du ser även antalet inkommande och utgående meddelanden på den här sidan. Du kan också se annan information, till exempel den aktuella storleken på kön, maximal storlek, antal aktiva meddelanden och så vidare.

Nästa steg

Se följande dokumentation och exempel: