问题标签 [parallel.foreach]

我刚开始使用 TPL，我想同时调用多个 Web 服务。据我所知，我看到了两种方法。

要么Parallel.ForEach：

或者Task<T>：

不管语法是否正确，这些是等价的吗？

它们会产生相同的结果吗？如果不是，为什么？哪个更可取？

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

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

我有一个 foreach 循环，里面有 Linq 查询。

在我将 foreach 更改为 Parallel.ForEach 之前，一切都运行良好：

我得到的例外是：

System.AccessViolationException 未处理 Message=尝试读取或写入受保护的内存。这通常表明其他内存已损坏。Source=Oracle.DataAccess StackTrace：在 Oracle.DataAccess.Client.ConnectionDispenser.Open(OpoConCtx opoConCtx) 的 Oracle.DataAccess.Client.ConnectionDispenser.Open(IntPtr& opsConCtx, IntPtr& opsErrCtx, OpoConValCtx* pOpoConValCtx, OpoConRefCtx& pOpoConRefCtx) 处。 System.Data.EntityClient.EntityConnection.OpenStoreConnectionIf 处的 Client.OracleConnection.Open()（布尔 openCondition，DbConnection storeConnectionToOpen，DbConnection originalConnection，字符串异常代码，字符串尝试操作，布尔和 closeStoreConnectionOnFailure）在 System.Data.EntityClient.EntityConnection.Open() 处。数据。1.GetResults(Nullable1 forMergeOption) 在 System.Data.Objects.ObjectQuery 1.System.Collections.Generic.IEnumerable<T>.GetEnumerator() at System.Collections.Generic.List1..ctor(IEnumerable 1 collection) at System.Linq.Enumerable.ToList[TSource](IEnumerable1 源) 在 ChangeOrder.Program.<>c_ DisplayClass19.b _16(SF_CHANGE_ORDER_ITEM t) 在 C:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program .cs：System.Threading.Tasks.Parallel.<>c_DisplayClass2d 处的 220 行1.b 2.<ForEachWorker>b__23(Int32 i) at System.Threading.Tasks.Parallel.<>c__DisplayClassf_c () 处 System.Threading.Tasks.Task.InnerInvoke() 处 System.Threading.Tasks.Task.InnerInvokeWithArg(Task childTask)在 System.Threading.Tasks.Task.<> c_DisplayClass7.b_6(Object ) 在 System.Threading.Tasks.Task.ExecuteSelfReplicating(Task root) 在 System.Threading.Tasks.Task.Execute() 在 System.Threading.Tasks.Task.ExecutionContextCallback(Object obj) 在 System.Threading.ExecutionContext .Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean ignoreSyncCtx) 在 System.Threading.Tasks.Task.ExecuteEntry(Boolean bPreventDoubleExecution) 在 System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot) 在 System.Threading.Tasks。 System.Threading.Tasks.TaskScheduler.TryRunInline(Task task, Boolean taskWasPrevouslyQueued, Object threadStatics) 在 System.Threading.Tasks.Task 的 ThreadPoolTask​​Scheduler.TryExecuteTaskInline(Task task, Boolean taskWasPreviousQueued)。InternalRunSynchronously(TaskScheduler scheduler) at System.Threading.Tasks.Task.RunSynchronously(TaskScheduler scheduler) at System.Threading.Tasks.Parallel.ForWorker[TLocal](Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Action1 body, Action2 bodyWithState，Func 4 bodyWithLocal, Func1 localInit，Action 1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IList1 列表，ParallelOptions parallelOptions，Action 1 body, Action2 bodyWithState，Action 3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal，Func 5 bodyWithEverything, Func1 localInit，Action 1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IEnumerable1 source，ParallelOptions，Action 1 body, Action2 bodyWithState，Action 3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal，Func 5 bodyWithEverything, Func1 localInit，Action1 localFinally) at System.Threading.Tasks.Parallel.ForEach[TSource](IEnumerable1 来源，Action`1 body) 在 ChangeOrder.Program.PerformChangeOrder(SF_CHANGE_ORDER co, SF_CLIENT_PROJECT cp, SFEntitiesQA ctx) 在 C:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program.cs:line 216 at ChangeOrder.Program.Main( C:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program.cs 中的 String[] args）：System.AppDomain._nExecuteAssembly 的第 1373 行（RuntimeAssembly 程序集，String[] args）在 System.AppDomain.ExecuteAssembly（字符串 assemblyFile，Evidence assemblySecurity, String[] args) 在 Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly() 在 System.Threading.ThreadHelper.ThreadStart_Context(Object state) 在 System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state,Boolean ignoreSyncCtx) 在 System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state) 在 System.Threading.ThreadHelper.ThreadStart() InnerException:

我不确定我需要锁定什么，因为我只是在获取信息（只是 RO，对吗？）。

我考虑过添加“AsParallel”，但我的理解是这是一个 PLINQ 指令，它只会导致查询相对于自身并行运行。

我找不到任何人在 Parallel.ForEach 循环中运行 Linq 查询的例子，所以我什至不确定我在做什么是允许的。

我将从对我如何理解几件事情的基本解释开始这一点，然后用一个 tldr 来结束这一切；如果人们只是想解决我在这里遇到的实际问题。如果我对这里的任何理解有误，请纠正我。

TPL 代表任务并行库，它是 .NET 4.0 试图进一步简化线程以方便开发人员使用的答案。如果您不熟悉它，（在非常基础的级别上）您启动一个新的 Task 对象并向其传递一个委托，然后该委托在从线程池获取的后台线程上运行（通过使用线程池而不是真正制作通过使用这些现有线程而不是创建和处置新线程来节省新线程、时间和资源）。

据我了解，C# 中的 Parallel.ForEach 命令将为它应该执行的每个委托生成一个新线程（可能来自线程池），但可能的例外是自动执行一个甚至可能更多的内联如果编译器认为它们将足够快地发生以提高效率，则迭代。

与我的目标最相关的背景信息：

我正在尝试制作一个启动任务的快速程序，以便与程序的其余部分同时运行。在此任务中，Parallel.ForEach 运行 3 次“迭代”。总的来说，我们希望程序现在总共运行 5 个线程（最多）：1 个用于主线程，1 个用于实际任务，最多 3 个用于 Parallel.ForEach。每个 Thread 都有自己的目标要完成（尽管 Parallel.ForEach 都有相同的目标，但其相关 itemNumber 的值不同以计算。当主线程完成所有目标时，它使用 Task.Wait() 等待在完成任务上，它也等待 Parallel.ForEach 完成。然后使用并验证这些值。

tldr; 实际问题：

运行上述想法时，Parallel.ForEach 似乎正在初始化两倍于我预期的 SynchronizationContexts（本质上是另一个线程的 TPL 对象）并运行所有它们，但只等待预期数量的它们。因为 Parallel.ForEach().Wait() 命令以预期的线程运行数量完成，所以 Task 也会在它认为一切都完成时完成。然后主程序发现任务已经完成，并且当它验证当前没有更多后台线程在运行时，有时剩余的 Parrallel.ForEach() 尚未完成，因此会引发错误。

通过在每个 SynchronizationContext 的 post 调用（Async 方法启动器）时打印到调试窗口来验证线程的数量与我所说的匹配。每个线程还被一个主线程对象引用，否则该对象计划在完成任务时被处置，但由于由于未真正预期创建的未完成线程仍然有引用，因此处置不能正确发生。

在上面的示例中，主线程、backgroundTask 任务和 8 个 Parallel.ForEach 线程最终都存在，其中最后 9 个线程是在 SynchronizationContexts 上创建的。

我的自定义在 SynchronizationContext 中覆盖的唯一方法是 post，如下所示：

我认为 TaskScheduler 只做它所需的基本工作：

任务工厂：

关于为什么会发生这种情况的任何想法？

今天我测试了 Linq 和 Plinq 查询的性能影响。因此，我使用了有关msdn How to: Measure PLINQ Query Performance的文章。

测试环境：LinqPad4 on Win7 Enterprise, 64bit, 8GB RAM, I7-2600 (8cores)

我想通了，但无法解释为什么一个核心上的查询（测量 1）比并行查询快。我是否必须添加更多选择代表才能从并行任务中受益？

但现在结果：

1.Run：可枚举范围为60000：

2.Run：可枚举范围为600000000：

有趣的事实：在执行测试脚本之前使用垃圾收集器会大大增加测量 4 的时间：

3.运行：可枚举范围为 600000000 和垃圾收集器（来自 LinqPad）：

In conclusion, can I say that method 1 is the most suitable option to perform small select queries and method 5 when the select delegates will increase?

我正在使用任务并行库Parallel.ForEach()来执行一些后台任务。我遇到了 Phil Haack 的 WebBackGrounder 应用程序。我想实现它，但只支持DoWork方法而不支持动作。关于如何在上下文中使用它的任何想法都将非常有用。否则我应该编写自己的类来处理这些操作。JobHostThreading.TaskParallel.ForEach

我的并行每种方法如下：

我有一个线程，称之为“解析线程”。

我在这个线程上执行了一些计算，其中最后一个涉及更多的并行计算。

问题是，当这些非常长且计算量大的异步 foreach 操作完成（约 30 分钟）时，“解析线程”不会恢复。我的 GUI 仍然响应，但是应该继续在“解析线程”上运行的 RaiseDirectoryParsed 函数从未被调用。到目前为止，我已经调试了程序，并且对于在这种情况下该怎么做感到非常困惑。

我写了一个小程序，将一堆文件转换为 pdf。

该程序执行以下操作：

• FileInfo从文件夹中获取对象数组 （10'000 文档）
• 对于每个FileInfo
  • FileInfo.CopyTo()使用,创建备份副本
  • 使用一些 Aspose 库将文档转换为 PDF
  • 转换后，将 PDF 复制到新的目的地
  • 在 foreach 内部，一个事件由 WinForm UI 引发和处理以显示一些进度

根据文档的大小，文档的转换可能需要 0-3 秒。我认为这将是一个完美的候选人Parallel.ForEach，所以我修改了程序。

然而，转换需要 1 小时而不是传统的 foreach 1.5 小时Parallel.Foreach（我尝试过的服务器有 2 x Intel Xeon Procs）。

我做错了什么或者我需要考虑什么才能获得更好的性能？

我有点困惑Parallel.ForEach。
它是什么Parallel.ForEach以及它到底做了什么？
请不要引用任何 MSDN 链接。

这是一个简单的例子：

我怎样才能用 重写这个例子Parallel.ForEach？

我正在尝试运行具有 32 个线程的并行 foreach 循环。

这个 foreach 循环正在调用：

并行 foreach 循环由大约 65000 个项目的列表组成。

在某些时候，在大约 6000 - 6700 个项目处，调用 WorkflowInvoker.Invoke(new MyWorkflow(), myInputDictionary) 会引发 AccessViolationException 并显示以下消息：“尝试读取或写入受保护的内存。这通常表明其他内存已损坏。”

我尝试了不同的方法，例如创建 WorkflowInvoker 的实例并将 MyWorkflow 的实例直接传递给新创建的 WorkflowInvoker，然后在类上调用 Invoke 方法。也只是将 WorkflowInvoker.Invoke 作为静态方法调用，但这并没有什么不同。应用程序每次都在同一点崩溃。由于线程的原因，很难调试和查明导致问题的确切原因。

当然，如果我在正常的 foreach 循环中进行调用，则不会出现问题。

起初我们认为虚拟机内存不足，因为应用程序开始消耗所有可用内存，直到达到总内存 (4GB) 的 85% 的限制，然后它似乎崩溃了。所以我们将内存大小更改为 1GB，看看我们是否可以在循环的早期引发异常，但是 Windows 开始进行垃圾收集，所以最后异常仍然发生在大约 6000 - 6700 个项目的同一点。

非常感谢任何关于我下一步应该采取什么行动的建议；-)

-- 编辑 - 添加了完整的堆栈跟踪 --

-- 额外的编辑 - 添加了有问题的方法 - 对其进行了一些混淆以删除业务逻辑名称 -