在 Parallel.ForEach 中嵌套 await

小编典典

在 Parallel.ForEach 中嵌套 await

all

在 Metro 应用程序中，我需要执行多个 WCF 调用。需要进行大量调用，因此我需要在并行循环中进行调用。问题是并行循环在 WCF 调用全部完成之前退出。

您将如何重构它以按预期工作？

var ids = new List<string>() { "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" };
var customers = new  System.Collections.Concurrent.BlockingCollection<Customer>();

Parallel.ForEach(ids, async i =>
{
    ICustomerRepo repo = new CustomerRepo();
    var cust = await repo.GetCustomer(i);
    customers.Add(cust);
});

foreach ( var customer in customers )
{
    Console.WriteLine(customer.ID);
}

Console.ReadKey();

阅读 68

2022-06-13

共1个答案

小编典典

背后的整个想法Parallel.ForEach()是你有一组线程，每个线程处理集合的一部分。正如您所注意到的，这不适用于async-
await，您希望在异步调用期间释放线程。

你可以通过阻塞ForEach()线程来“解决”这个问题，但这会破坏async-的全部意义await。

您可以做的是使用TPL Dataflow而不是Parallel.ForEach()，它也支持异步Tasks。

具体来说，您的代码可以使用 a 编写，使用lambdaTransformBlock将每个 id 转换为 a
。该块可以配置为并行执行。您可以将该块链接到一个将每个块写入控制台的块。设置块网络后，您可以将每个 id
设置为.Customer``async``ActionBlock``Customer``Post()``TransformBlock

在代码中：

var ids = new List<string> { "1", "2", "3", "4", "5", "6", "7", "8", "9", "10" };

var getCustomerBlock = new TransformBlock<string, Customer>(
    async i =>
    {
        ICustomerRepo repo = new CustomerRepo();
        return await repo.GetCustomer(i);
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
    });
var writeCustomerBlock = new ActionBlock<Customer>(c => Console.WriteLine(c.ID));
getCustomerBlock.LinkTo(
    writeCustomerBlock, new DataflowLinkOptions
    {
        PropagateCompletion = true
    });

foreach (var id in ids)
    getCustomerBlock.Post(id);

getCustomerBlock.Complete();
writeCustomerBlock.Completion.Wait();

尽管您可能希望将的并行性限制TransformBlock为一些小常数。此外，您可以限制
的容量TransformBlock并使用异步将项目添加到其中SendAsync()，例如如果集合太大。

与您的代码（如果有效）相比，另一个好处是，一旦完成单个项目，就会开始写入，而不是等到所有处理完成。

2022-06-13