并行运行异步方法 8 次

Question

如何将以下内容变成 Parallel.ForEach？

public async void getThreadContents(String[] threads)
{
    HttpClient client = new HttpClient();
    List<String> usernames = new List<String>();
    int i = 0;

    foreach (String url in threads)
    {
        i++;
        progressLabel.Text = "Scanning thread " + i.ToString() + "/" + threads.Count<String>();
        HttpResponseMessage response = await client.GetAsync(url);
        String content = await response.Content.ReadAsStringAsync();
        String user;
        Predicate<String> userPredicate;
        foreach (Match match in regex.Matches(content))
        {
            user = match.Groups[1].ToString();
            userPredicate = (String x) => x == user;
            if (usernames.Find(userPredicate) != user)
            {
                usernames.Add(match.Groups[1].ToString());
            }
        }
        progressBar1.PerformStep();
    }
}

我在假设异步和并行处理相同的情况下对其进行编码，但我刚刚意识到事实并非如此。我查看了我能找到的所有问题，但我似乎真的找不到适合我的例子。他们中的大多数缺乏可读的变量名。使用不解释它们包含的内容的单字母变量名称是陈述示例的可怕方式。

我通常在名为线程的数组中有 300 到 2000 个条目（包含论坛线程的 URL），并且似乎并行处理（由于许多 HTTP 请求）会加快执行速度。

在我可以使用 Parallel.ForEach 之前，我是否必须删除所有异步（我在 foreach 之外没有任何异步，只有变量定义）？我该怎么做呢？我可以在不阻塞主线程的情况下这样做吗？

顺便说一下，我使用的是 .NET 4.5。

Answer 1

我假设异步和并行处理是相同的

异步处理和并行处理是完全不同的。如果您不了解其中的区别，我认为您应该先阅读更多相关信息（例如what is the relation between Asynchronous and parallel programming in c#?）。

现在，您想要做的实际上并不那么简单，因为您想要异步处理一个大集合，并具有特定程度的并行性 (8)。对于同步处理，您可以使用Parallel.ForEach()（与ParallelOptions 一起配置并行度），但没有简单的替代方法可以与async 一起使用。

在您的代码中，这很复杂，因为您希望所有内容都在 UI 线程上执行。 （尽管理想情况下，您不应该直接从计算中访问 UI。相反，您应该使用 IProgress，这意味着代码不再需要在 UI 线程上执行。）

在 .Net 4.5 中执行此操作的最佳方法可能是使用 TPL 数据流。它的ActionBlock 完全符合您的要求，但它可能非常冗长（因为它比您需要的更灵活）。所以创建一个辅助方法是有意义的：

public static Task AsyncParallelForEach<T>(
    IEnumerable<T> source, Func<T, Task> body,
    int maxDegreeOfParallelism = DataflowBlockOptions.Unbounded,
    TaskScheduler scheduler = null)
{
    var options = new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = maxDegreeOfParallelism
    };
    if (scheduler != null)
        options.TaskScheduler = scheduler;

    var block = new ActionBlock<T>(body, options);

    foreach (var item in source)
        block.Post(item);

    block.Complete();
    return block.Completion;
}

在你的情况下，你会这样使用它：

await AsyncParallelForEach(
    threads, async url => await DownloadUrl(url), 8,
    TaskScheduler.FromCurrentSynchronizationContext());

这里，DownloadUrl() 是处理单个 URL（循环体）的 async Task 方法，8 是并行度（在实际代码中可能不应该是字面常量）和 @ 987654335@ 确保代码在 UI 线程上执行。

Answer 2

Stephen Toub 有一个good blog post on implementing a ForEachAsync。 Svick 的回答对于 Dataflow 可用的平台来说相当不错。

这是一个替代方案，使用来自 TPL 的分区器：

public static Task ForEachAsync<T>(this IEnumerable<T> source,
    int degreeOfParallelism, Func<T, Task> body)
{
  var partitions = Partitioner.Create(source).GetPartitions(degreeOfParallelism);
  var tasks = partitions.Select(async partition =>
  {
    using (partition) 
      while (partition.MoveNext()) 
        await body(partition.Current); 
  });
  return Task.WhenAll(tasks);
}

然后你可以这样使用它：

public async Task getThreadContentsAsync(String[] threads)
{
  HttpClient client = new HttpClient();
  ConcurrentDictionary<String, object> usernames = new ConcurrentDictionary<String, object>();

  await threads.ForEachAsync(8, async url =>
  {
    HttpResponseMessage response = await client.GetAsync(url);
    String content = await response.Content.ReadAsStringAsync();
    String user;
    foreach (Match match in regex.Matches(content))
    {
      user = match.Groups[1].ToString();
      usernames.TryAdd(user, null);
    }
    progressBar1.PerformStep();
  });
}

Answer 3

另一种选择是使用SemaphoreSlim 或AsyncSemaphore（其中is included in my AsyncEx library 并支持比SemaphoreSlim 更多的平台）：

public async Task getThreadContentsAsync(String[] threads)
{
  SemaphoreSlim semaphore = new SemaphoreSlim(8);
  HttpClient client = new HttpClient();
  ConcurrentDictionary<String, object> usernames = new ConcurrentDictionary<String, object>();

  await Task.WhenAll(threads.Select(async url =>
  {
    await semaphore.WaitAsync();
    try
    {
      HttpResponseMessage response = await client.GetAsync(url);
      String content = await response.Content.ReadAsStringAsync();
      String user;
      foreach (Match match in regex.Matches(content))
      {
        user = match.Groups[1].ToString();
        usernames.TryAdd(user, null);
      }
      progressBar1.PerformStep();
    }
    finally
    {
      semaphore.Release();
    }
  }));
}

Answer 4

你可以试试AsyncEnumerator NuGet Package的ParallelForEachAsync扩展方法：

using System.Collections.Async;

public async void getThreadContents(String[] threads)
{
    HttpClient client = new HttpClient();
    List<String> usernames = new List<String>();
    int i = 0;

    await threads.ParallelForEachAsync(async url =>
    {
        i++;
        progressLabel.Text = "Scanning thread " + i.ToString() + "/" + threads.Count<String>();
        HttpResponseMessage response = await client.GetAsync(url);
        String content = await response.Content.ReadAsStringAsync();
        String user;
        Predicate<String> userPredicate;
        foreach (Match match in regex.Matches(content))
        {
            user = match.Groups[1].ToString();
            userPredicate = (String x) => x == user;
            if (usernames.Find(userPredicate) != user)
            {
                usernames.Add(match.Groups[1].ToString());
            }
        }

        // THIS CALL MUST BE THREAD-SAFE!
        progressBar1.PerformStep();
    },
    maxDegreeOfParallelism: 8);
}