如何查看 Azure 服务总线队列中的所有消息？

Posted 2023-04-13

【中文标题】如何查看 Azure 服务总线队列中的所有消息？

【英文标题】：How to peek all messages in Azure Service Bus queue?

【发布时间】：2021-06-23 03:19:13

【问题描述】：

我想查看来自多个 Azure 服务总线队列的所有消息。之后，我想在 queueName、insertDate 之后过滤它们，并有机会在正文上进行全文搜索。

目前，我正在使用Microsoft.Azure.ServiceBus 包创建一个ManagementClient 来收集队列信息，然后使用MessageReceiver 来查看消息。

var managementClient = new ManagementClient(connectionString);

var queue = await managementClient.GetQueueRuntimeInfoAsync(queueName);

var count = queue.MessageCount;

var receiver = new MessageReceiver(connectionString, queueName);

var messagesOfQueue = new List<Message>();

for (var i = 1; i <= count; i++)

   messagesOfQueue.Add(await receiver.PeekAsync());

有没有更好的方法来获取所有消息？或者有没有办法只查看适用于过滤器的消息？

我还尝试使用 WindowsAzure.ServiceBus 包中的 QueueClient.PeekBatch 方法。但是尽管我设置了正确的messageCount 参数，但该方法并没有返回所有消息。

还有包Azure.Messaging.ServiceBus...所有这些包是怎么回事？

那么我应该使用哪些包，以及基于某些过滤器查看队列消息的最佳方法是什么？

【参考方案1】：

我目前正在使用的解决方案按预期工作，如下所示：

var receiver = serviceBusClient.CreateReceiver(queueName);

var messagesOfQueue = new List<ServiceBusReceivedMessage>();
var previousSequenceNumber = -1L;
var sequenceNumber = 0L;

do

  var messageBatch = await receiver.PeekMessagesAsync(int.MaxValue, sequenceNumber);

  if (messageBatch.Count > 0)
  
    sequenceNumber = messageBatch[^1].SequenceNumber;

    if (sequenceNumber == previousSequenceNumber)
      break;

    messagesOfQueue.AddRange(messageBatch);

    previousSequenceNumber = sequenceNumber;
  
  else
  
    break;
  
 while (true);

它使用nuget包Azure.Messaging.ServiceBus。

【参考方案2】：

目前，您从接收者那里收到一条消息。更好的选择是使用MessageReceiver 的PeekBatchAsync(Int64, Int32) 方法批量接收消息。

这是执行此操作的示例代码（但未经测试）：

var messagesOfQueue = new List<Message>();
var sequenceNumber = 0;
var batchSize = 100;//number of messages to receive in a single call
do

    var messages = await receiver.PeekBatchAsync(sequenceNumber, batchSize);
    messagesOfQueue.AddRange(messages);
    if (messages.Count > 0)
    
        sequenceNumber = messages[messages.Count-1].SequenceNumber;
    
    else
    
        break;
    
 while (true);

【讨论】：

好的，但正如我所说，我已经尝试过 PeekBatch 方法，但无法查看一批中的所有消息。或者是消息的最大数量限制为 100，我必须多次调用 PeekBatch 方法？

我认为您尝试了旧版 SDK 中的 PeekBatch 消息。您可以使用最新的 SDK (Azure.Messaging.ServiceBus) 进行尝试吗？关于批量大小为 100，这只是我选择的一个随机数。是的，使用这种方法，您需要多次调用 PeekBatch 消息。

感谢您的帮助。根据您的评论，我想出了一个可行的解决方案。

【参考方案3】：

该解决方案避免了两次获取具有相同序列号的消息。

序列号单调递增。我已经测试了大多数情况，除了当 sequenceNumber 达到最大值 (Long.MaxValue) 时将其翻转为 0。

using Azure.Messaging.ServiceBus;

private static async Task<List<ServiceBusReceivedMessage>> PeekAllMessages(string serviceBusConnectionString, string queueName)

    var client = new ServiceBusClient(serviceBusConnectionString);
    var receiver = client.CreateReceiver(queueName);

    var messages = new List<ServiceBusReceivedMessage>();

    var batchSize = 20;
    var sequenceNumber = 0L;
    do
    
        var messageBatch = await receiver.PeekMessagesAsync(batchSize, sequenceNumber);

        if (messageBatch.Count <= 0)
        
            break;
        

        // Increasing the SequenceNumber by 1 to avoid getting the message with the same SequenceNumber twice
        sequenceNumber = messageBatch[^1].SequenceNumber + 1;

        messages.AddRange(messageBatch);

     while (true);

    return messages;