Azure ServiceBus &;异步-生存，还是毁灭

我在Azure上运行服务总线，每秒泵送大约10-100条消息。

最近我已经切换到.net 4.5，并兴奋地重构了所有的代码，使'async'和'await'在每行至少两次，以确保它'正确'完成:)

现在我想知道它实际上是更好还是更坏。如果您能看一下代码片段并让我知道您的想法。我特别担心，如果线程上下文切换不是给我更多的痛苦比好处，从所有的异步…(看着!dumpheap，这绝对是一个因素)

只是一点描述-我将发布2个方法-一个在ConcurrentQueue上做while循环，等待新消息，另一个方法一次发送一个消息。我还按照Azure博士的要求使用了瞬态故障处理模块。

发送循环(从头开始，等待新消息):

private async void SendingLoop()
    {
        try
        {
            await this.RecreateMessageFactory();
            this.loopSemaphore.Reset();
            Buffer<SendMessage> message = null;
            while (true)
            {
                if (this.cancel.Token.IsCancellationRequested)
                {
                    break;
                }
                this.semaphore.WaitOne();
                if (this.cancel.Token.IsCancellationRequested)
                {
                    break;
                }
                while (this.queue.TryDequeue(out message))
                {                       
                    try
                    {
                        using (message)
                        {
                            //only take send the latest message
                            if (!this.queue.IsEmpty)
                            {
                                this.Log.Debug("Skipping qeued message, Topic: " + message.Value.Topic);
                                continue;
                            }
                            else
                            {
                                if (this.Topic == null || this.Topic.Path != message.Value.Topic)
                                    await this.EnsureTopicExists(message.Value.Topic, this.cancel.Token);
                                if (this.cancel.Token.IsCancellationRequested)
                                    break;
                                await this.SendMessage(message, this.cancel.Token);
                            }
                        }
                    }
                    catch (OperationCanceledException)
                    {
                        break;
                    }
                    catch (Exception ex)
                    {
                        ex.LogError();
                    }
                }
            }
        }
        catch (OperationCanceledException)
        { }
        catch (Exception ex)
        {
            ex.LogError();
        }
        finally
        {
            if (this.loopSemaphore != null)
                this.loopSemaphore.Set();
        }
    }

发送消息:

private async Task SendMessage(Buffer<SendMessage> message, CancellationToken cancellationToken)
    {
        //this.Log.Debug("MessageBroadcaster.SendMessage to " + this.GetTopic());
        bool entityNotFound = false;
        if (this.MessageSender.IsClosed)
        {
            //this.Log.Debug("MessageBroadcaster.SendMessage MessageSender closed, recreating " + this.GetTopic());
            await this.EnsureMessageSender(cancellationToken);
        }
        try
        {
            await this.sendMessageRetryPolicy.ExecuteAsync(async () =>
            {
                message.Value.Body.Seek(0, SeekOrigin.Begin);
                using (var msg = new BrokeredMessage(message.Value.Body, false))
                {
                    await Task.Factory.FromAsync(this.MessageSender.BeginSend, this.MessageSender.EndSend, msg, null);
                }
            }, cancellationToken);
        }
        catch (MessagingEntityNotFoundException)
        {
            entityNotFound = true;                
        }
        catch (OperationCanceledException)
        { }
        catch (ObjectDisposedException)
        { }
        catch (Exception ex)
        {
            ex.LogError();
        }
        if (entityNotFound)
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                await this.EnsureTopicExists(message.Value.Topic, cancellationToken);
            }
        }
    }

上面的代码来自一个'Sender'类，它每秒发送1条消息。我有大约50-100个实例在任何给定的时间运行，所以它可能是相当多的线程。

顺便说一句，不要担心过多的ensureessagesender, RecreateMessageFactory, EnsureTopicExists，它们不经常被调用。

我不只是有一个后台线程通过消息队列工作和同步发送消息更好，提供所有我需要的是发送一个消息在同一时间，不担心异步的东西，避免开销随之而来。

请注意，通常发送一个消息到Azure服务总线是几毫秒的问题，它不是很昂贵。(除了有时它很慢，超时或服务总线后端有问题，它可能会挂起一段时间，试图发送东西)。

谢谢，很抱歉写了这么长一篇文章，

Stevo

解决方案

这个例子能解决我的情况吗?

static void Main(string[] args)
    {
        var broadcaster = new BufferBlock<int>(); //queue
        var cancel = new CancellationTokenSource();
        var run = Task.Run(async () =>
        {
            try
            {
                while (true)
                {
                    //check if we are not finished
                    if (cancel.IsCancellationRequested)
                        break;                       
                    //async wait until a value is available
                    var val = await broadcaster.ReceiveAsync(cancel.Token).ConfigureAwait(false);
                    int next = 0;
                    //greedy - eat up and ignore all the values but last
                    while (broadcaster.TryReceive(out next))
                    {
                        Console.WriteLine("Skipping " + val);
                        val = next;
                    }
                    //check if we are not finished
                    if (cancel.IsCancellationRequested)
                        break;
                    Console.WriteLine("Sending " + val);
                    //simulate sending delay
                    await Task.Delay(1000).ConfigureAwait(false); 
                    Console.WriteLine("Value sent " + val);                        
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex);
            }
        }, cancel.Token);
        //simulate sending messages. One every 200mls 
        for (int i = 0; i < 20; i++)
        {
            Console.WriteLine("Broadcasting " + i);
            broadcaster.Post(i);
            Thread.Sleep(200);
        }
        cancel.Cancel();
        run.Wait();
    }