问题标签 [parallel-extensions]

我目前正在使用.Net(Rx)的响应式扩展中的并行扩展。我相信它们也可以通过 .Net 4 beta 版本获得。

1）有什么方法可以确定实际使用了多少线程。我认为这与Environment.ProcessorCount（逻辑核心数）有关，但我想检查一下。

2）有没有办法配置你想使用多少线程？我注意到有一个ParallelOptions.MaxDegreeOfParallelism属性看起来很有希望，但它似乎默认为 -1（对线程数没有限制），我也不太确定是否可以为当前应用程序设置一次，而不是通过在每次调用Parallel.For()等。

我正在寻找 C# 中并行化（多线程）排序算法的简单实现，该算法可以对List<T>数组或数组进行操作，并且可能使用并行扩展，但这部分并不是绝对必要的。

编辑：Frank Krueger 提供了一个很好的答案，但是我希望将该示例转换为不使用 LINQ 的示例。另请注意，Parallel.Do()似乎已被Parallel.Invoke().

谢谢。

我已经非常努力地让它工作并且没有运气。如何获得并行扩展来运行具有两个输入参数的函数？我正在使用更新的版本，即带有 3.5 框架的 Reactive Extensions。

我需要获得扩展来运行act（或函数ProcessOrder），但无论我尝试什么，我都无法让它去做。

我曾经能够做到以下几点：

但是已经不支持了

我试图了解 BlockingCollection 在 .NET 4 上的新并行堆栈的上下文中的目的。

MSDN文档说：

BlockingCollection 用作 IProducerConsumerCollection 实例的包装器，允许从集合中删除尝试阻塞，直到可以删除数据。类似地，可以创建一个 BlockingCollection 来强制 IProducerConsumerCollection 中允许的数据元素数量的上限；然后可能会阻止对集合的添加尝试，直到有空间可用于存储添加的项目。

但是，当我查看一些 IProducerConsumerCollection 的实现时，比如 ConcurrentQueue，我看到它们提供了无锁、线程安全的实现。那么为什么需要 BlockingCollection 提供的锁机制呢？MSDN 中的所有示例都显示通过 BlockingCollection 包装器使用这些集合，直接使用这些集合有什么麻烦？使用 BlockingCollection 有什么好处？

问题

尽管我将在这里讨论的代码是用 F# 编写的，但它是基于 .NET 4 框架的，并不具体取决于 F# 的任何特殊性（至少看起来如此！）。

我的磁盘上有一些数据，我应该从网络更新，将最新版本保存到磁盘：

问题是对于loadAndSaveAndUpdate我的所有数据，我必须多次执行该函数：

每一步都会做

• 一些磁盘 IO，
• 一些数据处理，
• 一些网络 IO（可能有很多延迟），
• 更多的数据处理，
• 和一些磁盘 IO。

在某种程度上并行完成这不是很好吗？不幸的是，我的阅读和解析功能都不是“异步工作流就绪”。

我想出的第一个（不是很好）解决方案

任务

我做的第一件事是设置一个Task[]并启动它们：

然后我按 CTRL+ESC 只是为了查看它使用了多少线程。15, 17, ..., 35, ..., 170, ... 直到杀死应用程序！出了点问题。

平行线

我做了几乎同样的事情，但使用Parallel.ForEach(...)和结果是一样的：很多很多很多线程。

一个有效的解决方案……有点

然后我决定只启动n线程，Task.WaitAll(of them)然后是其他n，直到没有更多可用的任务。

这是可行的，但问题是当它完成处理n-1任务时，它会等待，等待，等待最后一个由于大量网络延迟而坚持阻塞的任务。不是很好！

那么，你将如何解决这个问题呢？我很乐意查看不同的解决方案，包括异步工作流（以及在这种情况下如何调整我的非异步函数）、并行扩展、奇怪的并行模式等。

谢谢。

我正在使用 Visual Studio 2010 Beta 2。

在 Parallel.For 循环中，我使用不同的参数值执行相同的方法。执行后处理的数据必须存储在数据库中。

但是我有一个例外帽子说我无法使用来自不同线程的相同数据上下文。

所以问题将是如何处理来自多个线程的数据上下文和 SubmitChanges()？

至少自 .NET 4.0 发布以来，微软似乎在支持并行和异步编程方面付出了很多努力，并且似乎出现了很多围绕此的 API 和库。尤其是最近到处都经常提到以下花哨的名字：

• 反应式框架，
• PLINQ（并行 LINQ），
• TPL（任务并行库）和
• 并行扩展。

现在它们似乎都是 Microsoft 产品，而且它们似乎都针对 .NET 的异步或并行编程场景。但目前尚不清楚它们中的每一个究竟是什么以及它们之间的关系。有些可能实际上是同一回事。

简而言之，任何人都可以直接说明什么是什么？

考虑以下简单的代码模式：

如果我想使用 Parallel Extensions(PE) 并行化它，我可以简单地替换 for 循环结构，如下所示：

但是，PE 将执行不必要的工作，将工作分配给 Foo 被证明为假的那些项目的线程。因此，我认为中间包装器/过滤 IEnumerable 在这里可能是一种合理的方法。你同意？如果是这样，实现这一目标的最简单方法是什么？（顺便说一句，我目前正在使用 C#2，所以我会感谢至少一个不使用 lambda 表达式等的示例。）

我有一个要并行处理的可观察集合，然后在过滤时观察处理后的值，最后订阅一个接收过滤值的处理程序。

我的示例在语法上是正确的并且编译得很好，当我运行代码时，Where会评估执行过滤的语句。但是没有数据通过订阅。如果我删除AsParallel以便通过常规完成处理，则IEnumerable数据会通过并且一切都按预期工作。

这是我的示例，对字符串进行了一些处理：

下一个奇怪的事情是，如果我使用TakeWhile运算符，在我看来，它在概念上类似于 Where，观察 ParallelQuery 按预期工作：

向订阅添加一些日志记录代码表明在ToObservable转换之前接收到数据，但不是在转换之后：

第 4 行的 lambda 中的断点被命中，而第 6 行的 lambda 中的断点从未被命中。

为什么将TakeWhile数据传递给订阅者而Where不会？

如果它很重要，我会在 Visual Studio 2010 RC 中使用针对 .Net 4.0 Framework Client Profile 的项目开发我的代码。

更新：根据@Sergeys 的回答，我重新设计了Where过滤器的位置。以下代码按预期工作：

processedStrings必须首先将初始的 observable转换为 enumerable 以使其并行化，然后将其转换回 observable 以订阅最终结果，这仍然感觉有点尴尬。

我有一个任务来执行HttpWebRequest使用

这显然会因 a 而失败WebException。对于调用者，我想返回一个Task<HttpResult>whereHttpResult是一个辅助类型来封装响应（或不封装响应）。在这种情况下，4xx 或 5xx 响应也不例外。

因此，我在请求任务中附加了两个延续。一个与TaskContinuationOptions OnlyOnRanToCompletion另一个与OnlyOnOnFaulted。然后将整个事情包裹在 aTask<HttpResult>中，以获取一个结果，无论哪一个 continuation 完成。

三个子任务（请求加上两个延续）中的每一个都是使用该AttachedToParent选项创建的。

但是当调用者等待返回的外部任务时，AggregateException会抛出请求失败。

我想在错误的继续中观察，WebException以便客户端代码可以查看结果。在 on fault continuation 中添加 aWait会引发错误，但围绕此问题的 try-catch 无济于事。也不查看该Exception属性（如“使用 Task.Exception 属性观察异常”部分提示此处）。

我可以安装一个UnobservedTaskException事件处理程序来过滤，但由于事件没有提供与故障任务的直接链接，这可能会在应用程序的这一部分之外进行交互，并且是一个大锤来破解坚果的情况。

给定一个故障实例，Task<T>是否有任何方法将其标记为“故障处理”？

简化代码：

和：

（HandleWebRequestSuccess最终会衍生出进一步的任务来获取响应的内容......）

客户端应该能够等待任务然后查看它的结果，而不会因为预期的和已经处理的错误而抛出它。