0

下面我有一个使用 Azure 服务总线队列的相当简单的 .NET 控制台应用程序。

正如您将看到的，我正在使用 Task.Factory 启动 25 个接收器任务，然后调用我的 APM 样式的 BeginMessageReceive 方法。然后，在 EndMessageReceive 结束时，我再次调用 BeginMessageReceive 以保持循环继续进行。

我的问题是我如何才能实现同样的事情，但使用递归任务并可能利用 C# 5.0 async/await 从 APM 样式的 BeginMessageReceive/EndMessageReceive 切换到 TPL/TAP 方法？

using System;
using System.Configuration;
using System.Threading.Tasks;
using Microsoft.ServiceBus.Messaging;


namespace ServiceBusConsumer
{
    class Program
    {
        private static QueueClient _queueClient;

        private static void Main(string[] args)
        {    
            var connectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];
            _queueClient = QueueClient.CreateFromConnectionString(connectionString, "MyQueue");

            for (var i = 0; i < 25; i++ )
            {
                Task.Factory.StartNew(BeginMessageReceive);
            }

            Console.WriteLine("Waiting for messages...");
            Console.ReadKey();

            _queueClient.Close();

        } //end private static void Main(string[] args)

        private static void BeginMessageReceive()
        {
            _queueClient.BeginReceive(TimeSpan.FromMinutes(5), EndMessageReceive, null);
        }

        private static void EndMessageReceive(IAsyncResult iar)
        {
            var message = _queueClient.EndReceive(iar);
            try
            {
                if (message != null)
                {
                    var msg = message.GetBody<string>();
                    Console.WriteLine("Message: " + msg);

                    if (_queueClient.Mode == ReceiveMode.PeekLock)
                    {
                        // Mark brokered message as completed at which point it's removed from the queue.
                        message.Complete();
                    }
                }
            }
            catch (Exception ex)
            {
                if (_queueClient.Mode == ReceiveMode.PeekLock)
                {
                    // unlock the message and make it available 
                    message.Abandon();
                }

                Console.WriteLine("Exception was thrown: " + ex.GetBaseException().Message);
            }
            finally
            {
                if (message != null)
                {
                    message.Dispose();
                }
            }
            BeginMessageReceive();
        }

    }
}

如果 MessageReceive 超时过期，用于递归调用自身的新修改代码：

private static async Task MessageReceiveAsync()
{
    while (true)
    {
        using (var message = await _queueClient.ReceiveAsync(TimeSpan.FromMinutes(5)))try
        {
            if (message != null)
            {               
                try
                {

                    var msg = message.GetBody<string>();
                    Console.WriteLine("Message: " + msg);

                    if (_queueClient.Mode == ReceiveMode.PeekLock)
                    {
                        // Mark brokered message as completed at which point it's removed from the queue.
                        await message.CompleteAsync();
                    }
                }
                catch (Exception ex)
                {
                    message.AbandonAsync();
                    Console.WriteLine("Exception was thrown: " + ex.GetBaseException().Message);
                }
            }
        }
    }
}
4

1 回答 1

4

看起来 Azure 客户端库仍未使用 TAP API 进行更新。不知道那里有什么阻碍...

无论如何，您可以使用创建自己的APM->TAP 包装器，TaskFactory.FromAsync如下所示：

public static class MyAzureExtensions
{
  public static Task<BrokeredMessage> ReceiveAsync(this QueueClient @this,
      TimeSpan serverWaitTime)
  {
    return Task<BrokeredMessage>.Factory.FromAsync(
        @this.BeginReceive, @this.EndReceive, serverWaitTime, null);
  }

  public static Task CompleteAsync(this BrokeredMessage @this)
  {
    return Task.Factory.FromAsync(@this.BeginComplete, @this.EndComplete, null);
  }
}

将 Azure 调用封装到 TAP-ready API 后，就可以这样使用它们：

private static void Main(string[] args)
{    
  var connectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];
  _queueClient = QueueClient.CreateFromConnectionString(connectionString, "MyQueue");

  for (var i = 0; i < 25; i++ )
    MyMessageReceiveAsync();

  Console.WriteLine("Waiting for messages...");
  Console.ReadKey();

  _queueClient.Close();
}

private static async Task MyMessageReceiveAsync()
{
  while (true)
  {
    using (var message = await _queueClient.ReceiveAsync(TimeSpan.FromMinutes(5)))
    {
      try
      {
        var msg = message.GetBody<string>();
        Console.WriteLine("Message: " + msg);

        if (_queueClient.Mode == ReceiveMode.PeekLock)
        {
          // Mark brokered message as completed at which point it's removed from the queue.
          await message.CompleteAsync();
        }
      }
      catch (Exception ex)
      {
        if (_queueClient.Mode == ReceiveMode.PeekLock)
        {
          // unlock the message and make it available 
          message.Abandon();
        }

        Console.WriteLine("Exception was thrown: " + ex.GetBaseException().Message);
      }
    }
  }
}

像这样使用的一个好处async是您不会不必要地占用线程池线程。原来用了25个线程来监听；我的示例不会使用任何线程来收听。在我的示例中，线程池线程唯一被绑定的时间是消息被放弃时（在错误处理分支中）。

与原始代码有一个主要的语义差异：如果 的QueueClient“接收”引发异常，则在原始代码中它将使进程崩溃；在我的示例中，该异常将被忽略。

于 2013-02-07T13:56:25.973 回答