c# - .net Core Parallel.ForEach 问题

Question

我已经为某些项目切换到 .net Core，但现在遇到 Parallel.ForEach 问题。在过去，我经常有一个 id 值列表，然后我会用它来发出网络请求以获得完整的数据。它看起来像这样:

Parallel.ForEach(myList, l =>
{
    // make web request using l.id 
    // process the data somehow
});

那么，在 .net Core 中，Web 请求必须全部标记为 await，这意味着 Parallel.ForEach 操作必须标记为 async。但是，将 Parallel.ForEach 操作标记为 async 意味着我们有一个会导致问题的 void async 方法。在我的特定情况下，这意味着响应返回到应用程序之前并行循环中的所有 Web 请求都已完成，这既尴尬又会导致错误。

问题:在这里使用 Parallel.ForEach 有哪些替代方案？

我找到的一个可能的解决方案是将并行循环包装在任务中并等待任务:

await Task.Run(() => Parallel.ForEach(myList, l =>
{
    // stuff here
}));

(在此处找到:Parallel.ForEach vs Task.Run and Task.WhenAll)

但是，这对我不起作用。当我使用它时，我仍然会在循环完成之前返回到应用程序。

另一种选择:

var tasks = new List<Task>();
foreach (var l in myList)
{
    tasks.Add(Task.Run(async () =>
    {
         // stuff here
    }));
}
await Task.WhenAll(tasks);

这似乎可行，但这是唯一的选择吗？似乎新的 .net Core 使 Parallel.ForEach 几乎毫无用处(至少在涉及嵌套网络调用时)。

如有任何帮助/建议，我们将不胜感激。

Answer 1

为什么 Parallel.ForEach 不适合这个任务在评论中解释:它是为 CPU 绑定(bind)(CPU 密集型)任务设计的。如果你将它用于 IO 绑定(bind)操作(比如发出 Web 请求)——你将浪费线程池线程在等待响应时被阻塞，没有任何好处。仍然可以使用它，但它不是这种情况下的最佳选择。

您需要的是使用异步 Web 请求方法(如 HttpWebRequest.GetResponseAsync)，但这里又出现了另一个问题 - 您不想一次执行所有 Web 请求(如另一个答案所建议的)。您的列表中可能有数千个 url (id)。因此，您可以使用为此设计的线程同步结构，例如 Semaphore。 Semaphore 就像队列 - 它允许 X 个线程通过，其余的应该等待，直到其中一个繁忙的线程完成它的工作(有点简化的描述)。这是一个例子:

static async Task ProcessUrls(string[] urls) {
    var tasks = new List<Task>();
    // semaphore, allow to run 10 tasks in parallel
    using (var semaphore = new SemaphoreSlim(10)) {
        foreach (var url in urls) {
            // await here until there is a room for this task
            await semaphore.WaitAsync();
            tasks.Add(MakeRequest(semaphore, url));
        }
        // await for the rest of tasks to complete
        await Task.WhenAll(tasks);
    }
}

private static async Task MakeRequest(SemaphoreSlim semaphore, string url) {
    try {
        var request = (HttpWebRequest) WebRequest.Create(url);

        using (var response = await request.GetResponseAsync().ConfigureAwait(false)) {
            // do something with response    
        }
    }
    catch (Exception ex) {
        // do something
    }
    finally {
        // don't forget to release
        semaphore.Release();
    }
}