Очистка очереди сервисного автобуса Azure за один раз

Question

Мы используем очередь служебной шины в нашем проекте.Нам нужна функциональность для удаления всех сообщений из очереди, когда администратор решит очистить очередь.Я искал в сети, но не смог найти ни одной функции, которая делает это внутри класса QueueClient.

Нужно ли мне выводить все сообщения одно за другим, а затем отмечать их как завершенные, чтобы очистить очередь, или она естьлучший способ?

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

BrokeredMessage brokeredMessage = queueClient.Receive();

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

Answer 1

Использование метода Receive() в цикле while, как вы, приведет к тому, что ваш код будет выполняться бесконечно, когда очередь пуста, так как метод Receive() будет ожидать появления другого сообщения в очереди.

Если вы хотите, чтобы это запускалось автоматически, попробуйте использовать метод Peek().

Например:

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

Вы можете снова упростить это с ReceiveMode.ReceiveAndDelete, как былоупомянутый hocho.

Answer 2

Использование:

• Оба подхода (от @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 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

Самый быстрый способ очистить очередь 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 5

Для Azure-ServiceBus-Queues существует ReceiveBatch -метод , который позволяет вам получать партию n -мессий одновременно.В сочетании с ReceiveMode.ReceiveAndDelete вы можете очистить очередь более эффективно.

Предупреждение Число n сообщений может быть возвращено, но это не гарантируется.Также существует ограничение на размер пакета сообщений 256K .

Answer 6

Существует простой метод Clear() для очистки всей очереди, если вы используете библиотеку WindowsAzure.Storage из nuget. Я использую класс Microsoft.Windows.Azure.Queue из этой библиотеки для управления очередью. В противном случае вы можете получить доступ через их API согласно их документации . Я не знаю, как долго этот метод находился в библиотеке Azure, и, вероятно, его не было, когда вопрос был задан изначально, но API REST восходит по крайней мере с 2014 года на этого сообщения обратной связи Azure

Полный код .NET для очистки очереди с помощью библиотеки Azure:

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();