在Parallel.ForEach中嵌套等待

Z时代
2024-01-10
分类：问答

在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();

回答：

背后的整个想法Parallel.ForEach()是，您有一组线程，每个线程处理集合的一部分。正如您所注意到的，这不适用于async-

await，您想在异步调用期间释放线程。

您可以通过阻塞ForEach()线程来“修复”该问题，但这使async- 的全部观点无效await。

您可以做的是使用TPL Dataflow代替Parallel.ForEach()，它Task很好地支持异步。

具体来说，您的代码可以使用a编写TransformBlock，该代码Customer使用asynclambda 将每个id转换为a

。可以将该块配置为并行执行。您可以将该块链接到，ActionBlock然后将每个块写入Customer控制台。设置区块网络后，您可以将Post()每个ID分配到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();