c# - 通过 Azure 服务总线死信队列上的 MessageId 获取消息

Question

我正在编写用于监视和修复 Azure 上主题订阅的死信队列中的消息的功能。

我可以使用_subscriptionClient.PeekBatch(10)获取消息列表进行显示；然而，当我们想要实际删除其中一条消息时，我们就陷入了困境。

工作流程是:

1. 获取所有消息的列表
2. 将特定消息(例如集合中的第 5 条消息)发送回原始队列进行重新处理
3. 将关联的死信消息标记为 .Complete()。

问题是，虽然我们有消息列表，但如果不先使用 .subscriptionClient.Receive() ，我们就无法对特定消息调用 .Complete() 。

由于您无法通过 MessageId 来 .Receive() 消息，这是否意味着我们必须像下面这样一条一条地循环消息？

public BrokeredMessage GetMessageById(string messageIdentifier)
{
    BrokeredMessage matchingMessage = null;
    var messageNotFound = true;

    var messagesToAbandon = new List<BrokeredMessage>();

    while (messageNotFound)
    {
        var message = _subscriptionClient.Receive();

        if (message == null)
        {
            throw new Exception("Could not find the message on the queue");
        }

        if (message.MessageId == messageIdentifier)
        {
            messageNotFound = false;
            matchingMessage = message;
        }
        else
        {
            messagesToAbandon.Add(message);
        }
    }

    // Unlock all messages that do not match the matching one received.
    foreach (var message in messagesToAbandon)
    {
        message.Abandon();
    }

    return matchingMessage;
}

我对这种方法的问题是:

• 它不可扩展；如果该消息是集合中的第 100 条消息怎么办？我们必须遍历 99 个并将每个标记为已放弃
• 如果有太多要放弃的项目，我们可能会失去对 matchingMessage 的锁定
• 这是一个长期运行的过程

我曾考虑过如果循环内不匹配则将每条消息标记为已放弃的想法；然而，这会带来我们无限循环相同项目的风险(因为 .Abandon() 将它们放回到队列中)。

有人找到解决这个问题的有效方法吗？也许将 .Defer() 与 .Receive(sequenceNumber) 一起使用？

Answer 1

您可以在此处使用 session ，为每条消息设置 session ID。稍后，当您浏览死信时，选择消息及其 session ID，使用 session ID 打开 session 并接收消息。

更新:死信队列是一种特殊的队列，不允许 session ，也无法使用序列号从死信队列接收消息，如果您想挑选一条消息，这是我找到的两个选项。

因此，在问题中描述的情况下应该做什么，从死信中取出所有消息，将某些消息重播到原始队列进行重新处理，将其余消息移动到另一个队列( ErrorQueue)并清除死信队列。我认为这会更合适，因为死信队列更多的是 ServicBus 处理错误/过期的内部队列。如果应用程序需要处理错误，将它们移动到应用程序特定队列将提供更大的灵 active ，并且不会重载系统队列(死信队列)