将每个异步结果存储在其自己的数组元素中

Question

假设我想从网站下载 1000 个食谱。这些网站最多接受 10 个并发连接。 每个配方都应该存储在一个数组中，在其对应的索引处。（我不想将数组发送到DownloadRecipe 方法。）

从技术上讲，我已经解决了这个问题，但我想知道是否有更简洁的方法来使用 async/await 或其他方法来实现它？

    static async Task MainAsync()
    {
        int recipeCount = 1000;
        int connectionCount = 10;
        string[] recipes = new string[recipeCount];
        Task<string>[] tasks = new Task<string>[connectionCount];
        int r = 0;

        while (r < recipeCount)
        {
            for (int t = 0; t < tasks.Length; t++)
            {
                tasks[t] = Task.Run(async () => recipes[r] = await DownloadRecipe(r));
                r++;
            }

            await Task.WhenAll(tasks);
        }
    }

    static async Task<string> DownloadRecipe(int index)
    {
        // ... await calls to download recipe
    }

此外，此解决方案不是最佳解决方案，因为在所有 10 次运行下载完成之前，它不会开始新的下载。有什么我们可以在那里改进而不会使代码过于臃肿的地方吗？线程池限制为 10 个线程？

Answer 1

有很多方法可以做到这一点。一种方法是使用ActionBlock，它使您可以相当轻松地访问MaxDegreeOfParallelism，并且可以很好地与async 方法配合使用

static async Task MainAsync()
{
   var recipeCount = 1000;
   var connectionCount = 10;
   var recipes = new string[recipeCount];

   async Task Action(int i) => recipes[i] = await DownloadRecipe(i);
   
   var processor = new ActionBlock<int>(Action, new ExecutionDataflowBlockOptions()
   {
      MaxDegreeOfParallelism = connectionCount,
      SingleProducerConstrained = true
   });

   for (var i = 0; i < recipeCount; i++)
      await processor.SendAsync(i);

   processor.Complete();
   await processor.Completion;
}

static async Task<string> DownloadRecipe(int index)
{
   ...
}

另一种方法可能是使用SemaphoreSlim

var slim = new SemaphoreSlim(connectionCount, connectionCount);

var tasks = Enumerable
   .Range(0, recipeCount)
   .Select(Selector);
   
async Task<string> Selector(int i)
{
   await slim.WaitAsync()
   try
   {
      return await DownloadRecipe(i)
   }
   finally
   {
      slim.Release();
   }
}

var recipes = await Task.WhenAll(tasks);

另一组方法是使用响应式扩展 (Rx)... 再次，有很多方法可以做到这一点，这只是一种 等待 方法（考虑到所有事情可能会更好)

var results = await Enumerable
        .Range(0, recipeCount)
        .ToObservable()
        .Select(i => Observable.FromAsync(() => DownloadRecipe(i)))
        .Merge(connectionCount)
        .ToArray()
        .ToTask();

Answer 2

拥有 10 个“池”的替代方法将“同时”加载数据。

您不需要使用单独的线程来包装 IO 操作。使用单独的线程进行 IO 操作只是浪费资源。
请注意，下载数据的线程什么都不做，只是等待响应。这就是async-await 方法非常方便的地方 - 我们可以发送多个请求而无需等待它们完成并且不会浪费线程。

static async Task MainAsync()
{
    var requests = Enumerable.Range(0, 1000).ToArray();
    var maxConnections = 10;
    var pools = requests
        .GroupBy(i => i % maxConnections)
        .Select(group => DownloadRecipesFor(group.ToArray()))
        .ToArray();

    await Task.WhenAll(pools);

    var recipes = pools.SelectMany(pool => pool.Result).ToArray();
}

static async Task<IEnumerable<string>> DownLoadRecipesFor(params int[] requests)
{
    var recipes = new List<string>();
    foreach (var request in requests)
    {
        var recipe = await DownloadRecipe(request);
        recipes.Add(recipe);
    }

    return recipes;
}

因为在池中（DownloadRecipesFor 方法），我们会一一下载结果 - 我们确保我们一直有不超过 10 个活动请求。

这比原来的更有效一点，因为我们不会等到 10 个任务完成后才开始下一个“一堆”。
这并不理想，因为如果最后一个“池”提前结束，那么其他“池”将无法获取下一个要处理的请求。

最终结果会有相应的索引，因为我们将按照我们创建它们的顺序处理“池”和内部请求。