用于错误处理的并行 I/O 和重试逻辑

Question

通常并行处理仅与 CPU 密集型操作相关。但是，PLINQ 使用 WithDegreeOfParallelism 扩展专门提供 IO 密集型支持。例如：

from site in new[]
{
    "www.albahari.com",
    "www.linqpad.net",
    "www.oreilly.com",
    "www.takeonit.com",
    "stackoverflow.com",
    "www.rebeccarey.com"  
}
.AsParallel().WithDegreeOfParallelism(6)
let p = new Ping().Send (site)
select new
{
    site,
    Result = p.Status,
    Time = p.RoundtripTime
}

但是如果supporting IO is the goal of WithDegreeOfParallelism，那么PLINQ如何进一步扩展或使用来实现IO操作典型的“重试”效果？那么“延迟”效应呢？

例如，如果通过 WCF 服务调用的 IO 引发 CommunicationException，我可能希望在继续下一个资源之前使用“3 次尝试”策略再次发出相同的请求。我可能还想在每次尝试之间等待一分钟。虽然我在每次尝试之间“等待”一分钟，但我不希望线程阻塞等待。

在这个MSDN article 中，作者从一个类似于我上面显示的查询开始。然后他转换查询，以便没有线程阻塞。不幸的是，他在此过程中失去了 WithDegreeOfParallelism。而且无论哪种方式，他都没有解决发生错误时“重试”的问题。

有人知道或知道这样做的巧妙方法吗？

我正在考虑制作一个特殊的 IEnumerable 包装器，它允许在 PLINQ 遍历集合时“重新插入”值。这确实会导致“重试”行为，但仍然不允许“重试之间的 1 分钟延迟”要求。我可以使用 Thread.Sleep() 创建 1 分钟的延迟，但我尽量不阻塞线程。

想法？

Answer 1

您linked 的文章实际上展示了如何通过使用基于Task 的API (DownloadDataTask) 来避免阻塞IO 绑定操作的线程：

但是，此代码仍有一些不理想的地方。 这项工作（发送下载请求和阻止）几乎需要 没有 CPU，但它是由 ThreadPool 线程完成的，因为我使用的是 默认调度程序。理想情况下，线程应该只用于 CPU-bound 工作（当确实有工作要做时）。

使用 PLINK / Task.Run/ Task.Factory.StartNew 进行基于 IO 的操作是一种反模式。 PLINQ（与Parallel.For 等相同）适用于 CPU 密集型计算工作，但为自然异步网络/IO 密集型操作分配和阻塞线程毫无意义，它根本不需要线程“飞行中”。按照您显示的示例代码，这将类似于new Ping().SendAsync(site)，返回Task。然后您可以执行await Task.WhenAll(tasks) 并处理错误。

请参阅 Stephen Cleary 的 "There Is No Thread"，以及他最近的 answer 解决并行 IO 的最大程度问题。最重要的是，合并重试逻辑非常容易，无需涉及任何线程（例如，this）。