一次性清除 Azure 服务总线队列

Question

我们在项目中使用服务总线队列。当管理员选择清除队列时，我们需要一个从队列中删除所有消息的功能。我在网上搜索，但在 QueueClient 类中找不到任何功能。

我是否必须将所有消息一一弹出，然后将它们标记为完成以清除队列，还是有更好的方法？

QueueClient queueClient = _messagingFactory.CreateQueueClient(
                              queueName, ReceiveMode.PeekLock);

BrokeredMessage brokeredMessage = queueClient.Receive();

while (brokeredMessage != null )
{
    brokeredMessage.Complete();
    brokeredMessage = queueClient.Receive();
}

Answer 1

您可以简单地从 azure 门户中执行此操作。

如果您不担心过去 14 天（默认）存储的队列消息并想立即清除队列，则将消息时间更改为 5 或 10 秒并等待 5-10 秒。

重要步骤：几秒钟后，尝试从队列中接收以强制更新。

所有队列消息将被清除。您可以再次重置原始值。默认为 14 天。

Answer 2

当我读到这个问题时，我首先想到的是为什么不删除并重新创建队列？这可能是一种方法。但如果不想这样做，您可以接收并删除每条消息，直到没有消息为止。

您可以将最新的 Azure 服务总线 .NET 库 Azure.Messaging.ServiceBus 用于这两种方法。

1. 使用AdministrationClient 删除并重新创建队列

using Azure.Messaging.ServiceBus.Administration;

ServiceBusAdministrationClient adminClient = new ServiceBusAdministrationClient("<connectionstring>");
await adminClient.DeleteQueueAsync("myqueue");          
await adminClient.CreateQueueAsync("myqueue");          

注意：ServiceBusAdministrationClient 用于对已现有服务总线命名空间的 CRUD 操作。如果您还需要创建命名空间的能力，请改用Microsoft.Azure.Management.ServiceBus 库。

1. 接收并删除直到没有消息

using Azure.Messaging.ServiceBus;

await using var client = new ServiceBusClient("<connectionstring>");
                
ServiceBusReceiver receiver = client.CreateReceiver("myqueue", 
    new ServiceBusReceiverOptions { ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete });

while ((await receiver.PeekMessageAsync()) != null)
{
    // receive in batches of 100 messages.
    await receiver.ReceiveMessagesAsync(100);
}

ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete 是不言自明的。默认接收模式是PeekLock，它实际上并没有从队列中删除消息——它告诉服务总线接收客户端想要明确地settle接收到的消息。锁定部分意味着竞争接收者在锁定期间无法访问消息。

Answer 3

结合使用ReceiveAndDelete、PrefetchCount、ReceiveBatchAsync 和一个简单的真值循环而不是使用 Peek，我得到了很好的结果。下面是 MessagingFactory 的示例：

var receiverFactory = MessagingFactory.CreateFromConnectionString("ConnString");
var receiver = receiverFactory.CreateMessageReceiver("QName", ReceiveMode.ReceiveAndDelete);
receiver.PrefetchCount = 300;

bool loop = true;
while (loop)
{
    var messages = await receiver.ReceiveBatchAsync(300, TimeSpan.FromSeconds(1));
    loop = messages.Any();
}

只需要WindowsAzure.ServiceBus Nuget 包。

Answer 4

如果您使用的是 nuget 中的 WindowsAzure.Storage 库，则有一个简单的 Clear() 方法可以清除整个队列。我使用该库中的Microsoft.Windows.Azure.Queue 类来管理队列。否则，您可以通过他们的 API per their documentation 访问。我不知道该方法在 Azure 库中存在多长时间，并且在最初提出问题时它可能不存在，但 REST API 至少可以追溯到 2014 年this Azure feedback post

使用 Azure 库清除队列的完整 .NET 代码是：

string connectionString = "YourAzureConnectionStringHere";
string queueName = "YourWebJobQueueName";
CloudStorageAccount storageAccount = CloudStorageAccount.Parse(connectionString);

// Create the queue client, then get a reference to queue
CloudQueueClient queueClient = storageAccount.CreateCloudQueueClient();
queue = queueClient.GetQueueReference(GetQueueName(queueName));

// Clear the entire queue
queue.Clear();

Answer 5

清理 Azure ServiceBus 队列的最快方法是设置一个非常短的DefaultMessageTimeToLive，等待几秒钟，尝试从队列接收强制更新，然后恢复初始的DefaultMessageTimeToLive。

您可以通过门户或代码来完成：

var namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
var queueDescription = _namespaceManager.GetQueue(queueName);
var queueClient = QueueClient.CreateFromConnectionString(connectionString, queueName, ReceiveMode.ReceiveAndDelete);

var dl = queueDescription.EnableDeadLetteringOnMessageExpiration;
var ttl = queueDescription.DefaultMessageTimeToLive;

queueDescription.EnableDeadLetteringOnMessageExpiration = false;
queueDescription.DefaultMessageTimeToLive = TimeSpan.FromSeconds(1);

Thread.Sleep(5000);
var dumy = queueClient.ReceiveBatch(200, TimeSpan.FromSeconds(1)).ToArray();

queueDescription.EnableDeadLetteringOnMessageExpiration = dl;
queueDescription.DefaultMessageTimeToLive = ttl;

Answer 6

使用：

• 两种方法（来自@ScottBrady 和@participant）
• 还有MessageReceiver 抽象

您可以编写清空服务总线队列或主题/订阅的方法：

MessageReceiver messageReceiver = ...
while (messageReceiver.Peek() != null)
{
    // Batch the receive operation
    var brokeredMessages = messageReceiver.ReceiveBatch(300);

    // Complete the messages
    var completeTasks = brokeredMessages.Select(m => Task.Run(() => m.Complete())).ToArray();

    // Wait for the tasks to complete. 
    Task.WaitAll(completeTasks);
}

Answer 7

对于 Azure-ServiceBus-Queues，有一个 ReceiveBatch-method，它允许您同时接收一批 n-消息。结合ReceiveMode.ReceiveAndDelete可以更高效的清队列。

警告可能会返回 n 条消息，但不能保证。 256K 的消息批量大小也有限制。

Answer 8

像您一样在 while 循环中使用 Receive() 方法将导致您的代码在队列为空后无限期运行，因为 Receive() 方法将等待另一条消息出现在队列中。

如果您希望它自动运行，请尝试使用Peek() 方法。

例如：

while (queueClient.Peek() != null)
{
    var brokeredMessage = queueClient.Receive();
    brokeredMessage.Complete();
}

您可以使用 hocho 提到的 ReceiveMode.ReceiveAndDelete 再次简化此操作。