在 Message.Abandon 之后 Azure 服务总线队列消息死信

Question

我正在试用 Azure 服务总线队列。我有以下代码：

队列发送：

string strConnectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];
        var namespaceManager = NamespaceManager.CreateFromConnectionString(strConnectionString);
        if (!namespaceManager.QueueExists("Test"))
        {
            QueueDescription qD = new QueueDescription("Test");
            qD.DefaultMessageTimeToLive = new TimeSpan(05, 00, 00);
            qD.LockDuration = new TimeSpan(00, 02, 30);
            qD.MaxSizeInMegabytes = 5120;                
            namespaceManager.CreateQueue(qD);              
        }
        if (namespaceManager.QueueExists("Test"))
        {
            QueueClient client = QueueClient.CreateFromConnectionString(strConnectionString, "Test", ReceiveMode.PeekLock);
            var qMessage = Console.ReadLine();
            using (MemoryStream strm = new MemoryStream(Encoding.UTF8.GetBytes(qMessage)))
            {
                BrokeredMessage bMsg = new BrokeredMessage(strm);
                bMsg.MessageId = Guid.NewGuid().ToString();
                bMsg.TimeToLive = new TimeSpan(05, 00, 00);
                client.Send(bMsg);
                Console.WriteLine("Message sent");
            }
        }
        Console.ReadLine();

接收代码：

 string strConnectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];
        var namespaceManager = NamespaceManager.CreateFromConnectionString(strConnectionString);
        if (namespaceManager.QueueExists("Test"))
        {
            QueueClient client = QueueClient.CreateFromConnectionString(strConnectionString, "Test",ReceiveMode.PeekLock);
            if (client != null)
            {
                OnMessageOptions options = new OnMessageOptions();
                options.AutoComplete = false;
                options.AutoRenewTimeout = TimeSpan.FromSeconds(31);                 


                client.OnMessage((message) =>
                    {                           
                        Console.WriteLine(message.State.ToString());
                        Console.WriteLine("Message Id: " + message.MessageId);
                        Stream stream = message.GetBody<Stream>();
                        StreamReader reader = new StreamReader(stream);
                        Console.WriteLine("Message: " + reader.ReadToEnd());                                
                        Console.WriteLine("***************");
                        message.Abandon();
                    });
                Console.ReadLine();
            }
        }

我看到，每当我调用 Abandon 时，消息都会变成 DeadLettered。我的假设是它应该被激活并且可以被另一个客户接收。

Answer 1

你对BrokeredMessage.Abandon Api的理解是正确的。它旨在放弃在消息上获取的 peek-lock（但不放弃消息本身），从而使其可供其他接收者获取消息。

以下是我们对 peek-locked 消息的不同状态的设想：

基础知识优先

“为什么”：如果客户需要竞争消费者（工作队列）语义 - 他们需要多个工作人员同时处理队列中的不同消息，并保证 Exactly-Once - 然后他们使用 ReceiveMode.PeekLock。在这个模型中，每个工作人员（队列接收者）都需要一种方法来将其当前消息（作业）的进度传达给其他工作人员。因此，brokeredMessage 提供了4 个函数来表达状态。

“什么”：

• 如果当前 Worker 成功处理了一条消息 - 调用 BrokeredMessage.Complete()
• 如果 BrokeredMessage 无法由当前工作程序处理，并且希望在另一个工作程序上重试处理 - 那么，放弃消息。但是，这里的问题是：假设有 2 个工作人员，他们每个人都认为另一个人可以处理此消息并调用 Abandon - 很快他们将进入一个无限循环重试处理该消息！因此，为了避免这种情况，我们提供了一个名为 MaxDeliveryCount on QueueDescription 的配置。此设置保护消息从队列传递到接收者的次数限制。 在上面的示例中，每次您收到（和放弃）消息时，ServiceBus 服务上的“deliveryCount”都会递增。当它达到 10 时 - 消息已达到最大值。交货，因此，将被死记。
• 如果当前接收者（工作人员）确定，该消息无法处理，BrokeredMessage.DeadLetter()。这里的目标是让消费应用程序定期审核死信消息。
• 如果当前接收者（工作人员）无法处理此消息，但知道此消息可以在稍后的时间点处理BrokeredMessage.Defer()。

HTH！ 斯里