如何使用Java将错误消息移动到Azure死信队列? [英] How to move error message to Azure dead letter queue using Java?

问题描述

我们正在使用Azure Service Bus队列在不同系统之间交换消息.我们想使用Java代码将无效消息移至Dead Letter Queue.

We are using Azure Service Bus Queues for exchanging messages between different systems. We would like to move the invalid messages to Dead Letter Queue using Java code.

我能够将邮件移至主队列，但不能移至死信队列.我试图将队列名称命名为"BasicQueue/$ deadletterqueue"，但出现错误.

I am able to move the messages to the Main queue, but not to dead letter queue. I tried to give the queue name as "BasicQueue/$deadletterqueue", but I was getting the error.

推荐答案

Azure Service总线运行时可能会将消息自动移动到死信队列的原因很多，例如:

There are many reasons why a message may be automatically moved to the dead letter queue by the Azure Service bus runtime, such as:

• 已超过最大投放次数(启用时默认为10)；
• 已超过消息时间(TTL)；
• 等等.

在中查找更多详细信息.https://docs.microsoft.com/zh-CN/azure/service-bus-messaging/service-bus-dead-letter-queues ，其中包括将邮件移至死信原因的完整列表队列.

Find more details at https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dead-letter-queues, including a full list of reasons why messages are moved to the dead letter queue.

此外，应用程序还可以将消息移至死信队列.在下面进一步找到示例 Java 8控制台应用程序代码和pom.xml文件.

In addition, it is possible for the application to move messages to the dead letter queue. Find a sample Java 8 Console Application code and the pom.xml file further below.

请注意，当将消息移至死信队列时，Azure Service Bus运行时会自动填充以下属性，最好在应用程序执行相同操作时也提供这些属性:

Please note The following properties are automatically filled up by the Azure Service Bus runtime when moving messages to the dead letter queue, it is a good idea to also provides those when the application it performing the same action:

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)

--- pom.xml的Azure服务总线依赖性---

<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 .

这篇关于如何使用Java将错误消息移动到Azure死信队列?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！