将阻塞调用包装为异步，以实现更好的线程重用和响应式 UI

我有一个类负责通过调用遗留类来检索产品可用性。该遗留类本身通过进行 BLOCKING 网络调用在内部收集产品数据。 请注意，我无法修改旧版 API 的代码。由于所有产品都是相互独立的，因此我希望并行收集信息，而不会创建任何不必要的线程，也不会阻塞在调用此旧版 API 时被阻塞的线程。有了这个背景，这是我的基础课程。

class Product
    {
        public int ID { get; set; }
        public int  VendorID { get; set; }
        public string Name { get; set; }
    }

    class ProductSearchResult
    {
        public int ID { get; set; }
        public int AvailableQuantity { get; set; }
        public DateTime ShipDate { get; set; }
        public bool Success { get; set; }
        public string Error { get; set; }
    }

class ProductProcessor
    {
        List<Product> products;
        private static readonly SemaphoreSlim mutex = new SemaphoreSlim(2);
        CancellationTokenSource cts = new CancellationTokenSource();
        public ProductProcessor()
        {
            products = new List<Product>()
            {
                new Product() { ID = 1, VendorID = 100, Name = "PC" },
                new Product() { ID = 2, VendorID = 101, Name = "Tablet" },
                new Product() { ID = 3, VendorID = 100, Name = "Laptop" },
                new Product() { ID = 4, VendorID = 102, Name = "GPS" },
                new Product() { ID = 5, VendorID = 107, Name = "Mars Rover" }
            };

        }

        public async void Start()
        {
            Task<ProductSearchResult>[] tasks = new Task<ProductSearchResult>[products.Count];
            Parallel.For(0, products.Count(), async i =>
            {
                tasks[i] = RetrieveProductAvailablity(products[i].ID, cts.Token);

            });



            Task<ProductSearchResult> results = await Task.WhenAny(tasks);

            // Logic for waiting on indiviaul tasks and reporting results

        }

        private async Task<ProductSearchResult> RetrieveProductAvailablity(int productId, CancellationToken cancellationToken)
        {
            ProductSearchResult result = new ProductSearchResult();
            result.ID = productId;

            if (cancellationToken.IsCancellationRequested)
            {
                result.Success = false;
                result.Error = "Cancelled.";
                return result;
            }

            try
            {
                await mutex.WaitAsync();
                if (cancellationToken.IsCancellationRequested)
                {
                    result.Success = false;
                    result.Error = "Cancelled.";
                    return result;
                }

                LegacyApp app = new LegacyApp();
                bool success = await Task.Run(() => app.RetrieveProductAvailability(productId));
                if (success)
                {
                    result.Success = success;
                    result.AvailableQuantity = app.AvailableQuantity;
                    result.ShipDate = app.ShipDate;
                }
                else
                {
                    result.Success = false;
                    result.Error = app.Error;
                }
            }
            finally
            {
                mutex.Release();
            }

            return result;

        }

    }

鉴于我正在尝试将异步封装在同步 API 上，我有两个问题。

1. 通过使用 Parallel.For 并将传统 API 调用包装在 Task.Run 中，我是否创建了任何不必要的线程，这些线程本可以在不阻塞调用线程的情况下避免，因为我们将在 UI 中使用此代码。
2. 这段代码看起来仍然是线程安全的吗？

编译器会给你关于你的警告async拉姆达。仔细阅读；它告诉你它不是异步的。使用没有意义async那里。另外，不要使用async void.

由于您的底层 API 是阻塞的 - 并且无法更改它 - 异步代码不是一个选项。我建议使用几个Task.Run calls or Parallel.For，但不能两者兼而有之。所以让我们使用并行。实际上，让我们使用并行 LINQ，因为您转变一个序列。

没有意义做RetrieveProductAvailablity异步;它只做除节流之外的阻塞工作，并且并行方法具有更自然的节流支持。这使得你的方法看起来像：

private ProductSearchResult RetrieveProductAvailablity(int productId, CancellationToken cancellationToken)
{
  ... // no mutex code
  LegacyApp app = new LegacyApp();
  bool success = app.RetrieveProductAvailability(productId);
  ... // no mutex code
}

然后您可以进行并行处理，如下所示：

public void Start()
{
  ProductSearchResult[] results = products.AsParallel().AsOrdered()
      .WithCancellation(cts.Token).WithDegreeOfParallelism(2)
      .Select(product => RetrieveProductAvailability(product.ID, cts.Token))
      .ToArray();
  // Logic for waiting on indiviaul tasks and reporting results
}

从你的 UI 线程中，你可以call该方法使用Task.Run:

async void MyUiEventHandler(...)
{
  await Task.Run(() => processor.Start());
}

这使您的业务逻辑保持干净（仅同步/并行代码），并负责将这项工作移出 UI 线程（使用Task.Run）属于UI层。

Update:我添加了一个电话AsOrdered确保结果数组的顺序与产品序列相同。这可能是必要的，也可能不是必要的，但由于原始代码保留了顺序，因此该代码现在也这样做了。

Update:由于您需要在每次检索后更新 UI，因此您可能应该使用Task.Run对于每一个而不是AsParallel:

public async Task Start()
{
  var tasks = products.Select(product =>
      ProcessAvailabilityAsync(product.ID, cts.Token));
  await Task.WhenAll(tasks);
}

private SemaphoreSlim mutex = new SempahoreSlim(2);
private async Task ProcessAvailabilityAsync(int id, CancellationToken token)
{
  await mutex.WaitAsync();
  try
  {
    var result = await RetrieveProductAvailability(id, token);
    // Logic for reporting results
  }
  finally
  {
    mutex.Release();
  }
}