Как переместить сообщение об ошибке в очередь недоставленных сообщений Azure с помощью Java?

Question

Мы используем очереди служебной шины Azure для обмена сообщениями между различными системами. Мы хотели бы переместить недопустимые сообщения в очередь недоставленных сообщений, используя код Java.

Я могу переместить сообщения в основную очередь, но не в очередь недоставленных сообщений. Я попытался дать имя очереди как «BasicQueue / $ deadletterqueue», но получаю ошибку.

Answer 1

Существует множество причин, по которым сообщение может автоматически перемещаться в очередь недоставленных сообщений средой выполнения шины службы Azure, например:

• Превышено максимальное количество доставок (по умолчанию 10, если включено);
• Превышено время сообщения (TTL);
• Etc;.

Более подробную информацию можно найти по адресу https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dead-letter-queues,, включая полный список причин, по которым сообщения перемещаются в очередь недоставленных сообщений.

Кроме того, приложение может перемещать сообщения в очередь недоставленных сообщений. Найдите пример кода консольного приложения Java 8 и файл pom.xml ниже.

Обратите внимание. Следующие свойства автоматически заполняются средой выполнения Azure Service Bus при перемещении сообщений в очередь недоставленных сообщений. Рекомендуется также указать их, когда приложение выполняет то же действие:

DeadLetterReason

DeadLetterErrorDescription

- Версия Java -

java version "1.8.0_162"
Java(TM) SE Runtime Environment (build 1.8.0_162-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.162-b12, mixed mode)

--- Зависимость служебной шины Azure для pom.xml ---

<dependencies>
    <!-- https://mvnrepository.com/artifact/com.microsoft.azure/azure-servicebus -->
    <dependency>
        <groupId>com.microsoft.azure</groupId>
        <artifactId>azure-servicebus</artifactId>
        <version>2.0.0-preview1</version>
    </dependency>
</dependencies>

- Консольное приложение Java 8 -

import java.time.Duration;
import java.util.Scanner;
import com.microsoft.azure.servicebus.MessageHandlerOptions;
import com.microsoft.azure.servicebus.QueueClient;
import com.microsoft.azure.servicebus.ReceiveMode;
import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
import com.microsoft.azure.servicebus.primitives.ServiceBusException;

public class Main {
    public static void main(String[] args)
    {
        final String CONNECTION_STRING = "{Azure Service Bus Connection String}";
        final String QUEUE_NAME = "{Azure Service Bus Queue Name}";

        try {
            moveMessageToDlq(CONNECTION_STRING, QUEUE_NAME);
        }catch (Exception ex) {
            System.out.println(String.format("An exception occurred. Details: %s.", ex.toString()));
        }
    }

    private static void moveMessageToDlq(String connectionString, String queueName)
            throws ServiceBusException, InterruptedException {

        ConnectionStringBuilder connectionStringBuilder = new ConnectionStringBuilder(connectionString, queueName);
        QueueClient queueClient = new QueueClient(connectionStringBuilder, ReceiveMode.PEEKLOCK);

        MessageHandler messageHandler = new MessageHandler(queueClient);

        MessageHandlerOptions messageHandlerOptions = new MessageHandlerOptions(
                1,
                false,
                Duration.ofMinutes(1));

        queueClient.registerMessageHandler(messageHandler, messageHandlerOptions);

        final String QUIT_COMMAND_NAME = "quit";
        Scanner scanner = new Scanner(System.in);

        while (true) {
            System.out.println(String.format("Enter '%s' and press <ENTER> to exit...", QUIT_COMMAND_NAME));
            String input = scanner.nextLine();

            if (input.equalsIgnoreCase(QUIT_COMMAND_NAME)){
                System.out.println("Exiting...");
                break;
            }
        }

        scanner.close();

        queueClient.close();
    }
}

--- Обработчик сообщений ---

import java.util.concurrent.CompletableFuture;
import com.microsoft.azure.servicebus.ExceptionPhase;
import com.microsoft.azure.servicebus.IMessage;
import com.microsoft.azure.servicebus.IMessageHandler;
import com.microsoft.azure.servicebus.IQueueClient;

public class MessageHandler implements IMessageHandler {
    private final IQueueClient _client;

    public MessageHandler(IQueueClient client) {
        if (client == null){
            throw new IllegalArgumentException("Queue client cannot be null.");
        }
        _client = client;
    }

    @Override
    public CompletableFuture<Void> onMessageAsync(IMessage message) {
        System.out.println(String.format("Received message id '%s' (DeliveryCount=%d).",
                message.getMessageId(),
                message.getDeliveryCount()));

        if (message.getLabel().equalsIgnoreCase("dlq")){ // Send message to DLQ if label id dlq
            System.out.println("Sending message to the dead letter queue...");

            return _client.deadLetterAsync(
                    message.getLockToken(),
                    "InvalidMessage", // DeadLetterReason
                    "Message invalid due to..."); // DeadLetterErrorDescription
        }
        else { // Otherwise, complete message
            System.out.println("Completing message...");

            return _client.completeAsync(message.getLockToken());
        }
    }

    @Override
    public void notifyException(Throwable throwable, ExceptionPhase exceptionPhase) {
        System.out.println("An exception occurred. Details: " + exceptionPhase + "-" + throwable.getMessage());
    }
}

---

Примеры использования GitHub Azure Service Bus Java: https://github.com/Azure/azure-service-bus-java/tree/e3d163eac92213d34dce059f3353d2c819d31fbf.