c# - 如何优雅地关闭Azure ServiceBus QueueClient？

Question

当我调用QueueClient.Close()时，它总是引发异常:

The operation cannot be performed because the entity has been closed or aborted.

即使队列为空，它也会引发异常。

虽然我使用OnMessageOptions.ExceptionReceived来处理它，但这让我很烦。我的代码有问题吗？

如何优雅地停止 QueueClient？

[启动和停止消息传递]

// create a QueueClient with Exception handler.
var queueClient = queueManager.GetStorage<UpdateTriggerQueueClient>();
var options = new OnMessageOptions
{
    AutoComplete = false
};
// When the Close() called, it always handles an exception.
options.ExceptionReceived += (sender, args) => 
    logger.Error("An excepion occurred.", args.Exception);

// prepare a CancellationTokenSource.
var cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = cancellationTokenSource.Token;

// start message pump.
queueClient.OnMessageAsync(async message => 
            await DoWork(message, cancellationToken), options);

// sometime after, stop(cancel) the work.
Task.Delay(5000).Wait();
cancellationTokenSource.Cancel();

// some code to wait every in-progress messages finished.
// ...

// close the client.
queueClient.Close();

[DoWork方法]

private async Task DoWork(BrokeredMessage message, CancellationToken cancellationToken)
{
    logger.Trace("begin work");
    // Do something cancellable work.
    await Task.Delay(500, cancellationToken)
        .ContinueWith(async t =>
        {
            // complete the message when the task completed,
            // otherwise, abandon the message.
            if (t.Status == TaskStatus.RanToCompletion)
            {
                await message.CompleteAsync();
            }
            else
            {
                await message.AbandonAsync();
            }
        })
        .ContinueWith(t =>
        {
            // cleanup
            logger.Trace("end work");
        });
}

Answer 1

您可以通过更新订阅来停止接收所有消息:

_namespaceManager.UpdateSubscription(new SubscriptionDescription(_strTopic, _strSubscription) 
  { 
     Status = EntityStatus.ReceiveDisabled
  });

这可能会解决您的问题，即使这并不完全是您所要求的。 (我也在尝试找出如何正确地 close() 。

您还需要更新状态才能再次开始接收。