在 Azure 服务总线上的死信队列中完成一条消息答案

【问题标题】：Complete a message in a dead letter queue on Azure Service Bus在 Azure 服务总线上的死信队列中完成一条消息
【发布时间】：2015-11-10 00:42:35
【问题描述】：

我希望能够从我的死信队列中删除选定的消息。

这是如何实现的？

我经常收到错误：

操作无法完成，因为 RecieveContext 为 Null

我已经尝试了所有我能想到和读到的方法，这就是我现在的位置：

public void DeleteMessageFromDeadletterQueue<T>(string queueName, long sequenceNumber)
        {
     var client = GetQueueClient(queueName, true);
                var messages = GetMessages(client);
                foreach(var m in messages)
                {
                    if(m.SequenceNumber == sequenceNumber)
                    {
                        m.Complete();
                    }
                    else
                    {
                        m.Abandon();
                    }
                }
}

/// <summary>
        /// Return a list of all messages in a Queue
        /// </summary>
        /// <param name="client"></param>
        /// <returns></returns>
        private IEnumerable<BrokeredMessage> GetMessages(QueueClient client)
        {
            var peekedMessages = client.PeekBatch(0, peekedMessageBatchCount).ToList();
            bool getmore = peekedMessages.Count() == peekedMessageBatchCount ? true : false;

            while (getmore)
            {
                var moreMessages = client.PeekBatch(peekedMessages.Last().SequenceNumber, peekedMessageBatchCount);
                peekedMessages.AddRange(moreMessages);
                getmore = moreMessages.Count() == peekedMessageBatchCount ? true : false;
            }

            return peekedMessages;
        }

不知道为什么这似乎是一项如此艰巨的任务。

【问题讨论】：

标签： azure azureservicebus azure-servicebus-queues

【解决方案1】：

这里的问题是您调用了 PeekBatch，它返回了刚刚被偷看的消息。没有接收上下文，您可以使用它来完成或放弃消息。 Peek 和 PeekBatch 操作只返回消息并且根本不锁定它们，即使 receivedMode 设置为 PeekLock。它主要用于浏览队列，但您无法对它们采取行动。请注意 Abandon 和 Complete 状态的文档，“只能在使用在 Peek-Lock ReceiveMode 下运行的接收器接收到的消息上调用。”这里不清楚，但 Peek 和 PeekBatch 操作不计入其中，因为它们实际上并没有获得接收上下文。这就是当您尝试调用放弃时它失败的原因。如果你真的找到了你正在寻找的那个，当你调用 Complete 时它会抛出一个不同的错误。

您想要做的是在 PeekBatch RecieveMode 中使用 ReceiveBatch 操作。这实际上会拉回一批消息，然后当您查看它们以找到您想要的消息时，您实际上可以影响消息的完成。当您触发放弃时，它会立即将您不想要的消息释放回队列。

如果您的死信队列非常小，通常这还不错。如果它真的很大，那么采用这种方法并不是最有效的。您将死信队列更像是一个堆并挖掘它，而不是“按顺序”处理消息。这在处理需要手动干预的死信队列时并不少见，但如果你有很多这样的东西，那么最好让一些东西将死信队列处理到不同类型的存储中，你可以更容易地找到和销毁消息，但仍然可以重新创建可以推送到不同队列以重新处理的消息。

可能还有其他选项，例如使用 Defer，如果您手动进行死字处理。见How to use the MessageReceiver.Receive method by sequenceNumber on ServiceBus。

【讨论】：

感谢您在这里指出我的错误，呃，现在我只需要一种方法来从队列中获取所有消息
我不认为你是这个意思，但你不想同时从队列中获取所有消息。这有点危险，因为队列中实际有多少是未知数。这真的是用于死信队列处理吗？你在死信队列中看到了什么样的数字？

【解决方案2】：

MikeWo 的建议没有成功，因为当我结合使用 ReceiveMode.PeekLock 实例化 DLQ QueueClient 和使用 ReceiveBatch 拉取消息时，我使用的 Receive/ReceiveBatch 版本通过其 SequenceNumber 请求消息。

[除此之外：在我的应用程序中，我查看所有消息并列出它们，并让另一个处理程序根据其特定序列号将死信消息重新排队到主队列...]

但在 DLQClient 上调用 Receive(long sequenceNumber) 或 ReceiveBatch(IEnumerable sequenceNumber) 总是会抛出异常“无法锁定一条或多条指定消息。消息不存在。” （即使我只通过了 1 并且它肯定在队列中）。

另外，由于不清楚的原因，使用 ReceiveBatch(int messageCount)，无论使用什么值作为 messageCount，总是只返回队列中的下一条消息。

最后对我有用的是：

QueueClient queueClient, deadLetterClient;
GetQueueClients(qname, ReceiveMode.PeekLock, out queueClient, out deadLetterClient);

BrokeredMessage msg = null;
var mapSequenceNumberToBrokeredMessage = new Dictionary<long, BrokeredMessage>();
while (msg == null)
{
#if UseReceive
    var message = deadLetterClient.Receive();
#elif UseReceiveBatch
    var messageEnumerable = deadLetterClient.ReceiveBatch(CnCountOfMessagesToPeek).ToList();
    if ((messageEnumerable == null) || (messageEnumerable.Count == 0))
        break;
    else if (messageEnumerable.Count != 1)
        throw new ApplicationException("Invalid expectation that there'd always be only 1 msg returned by ReceiveBatch");

    // I've yet to get back more than one in the deadletter queue, but... 
    var message = messageEnumerable.First();
#endif
    if (message.SequenceNumber == lMessageId)
    {
        msg = message;
        break;
    }
    else if (mapSequenceNumberToBrokeredMessage.ContainsKey(message.SequenceNumber))
    {
        // this means it's started the list over, so we didn't find it...
        break;
    }
    else
        mapSequenceNumberToBrokeredMessage.Add(message.SequenceNumber, message);
    message.Abandon();
}                        

if (msg == null)
    throw new ApplicationException("Unable to find a message in the deadletter queue with the SequenceNumber: " + msgid);

var strMessage = GetMessage(msg);
var newMsg = new BrokeredMessage(strMessage);
queueClient.Send(newMsg);

msg.Complete();

【讨论】：