Linq 并行.ForEach

Question

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

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

// get the task info   ---------
                Log("Populate task, guf code lists ...........................");
                List<SF_CO_ITEM> tasks = (from coi in ctx.SF_CO_ITEM
                                                    where coi.CO == co.ID
                                                    select coi).ToList();

               // foreach (SF_CO_ITEM t in tasks)
               // {
                Parallel.ForEach(tasks, t =>
                {
                    Log("Executing on t: " + t.ID);

                    // exception on next line:
                    List<SF_CO_LINE_ITEM> gufs = (from coli in ctx.SF_CO_LINE_ITEM      
                                                            where coli.CO_ITEM == t.ID
                                                            select coli).ToList();

我得到的例外是：

System.AccessViolationException 未处理 Message=Attempted to 读或写受保护的内存。这通常表明其他 内存已损坏。来源=Oracle.DataAccess StackTrace： 在 Oracle.DataAccess.Client.OpsCon.Open（IntPtr& opsConCtx，IntPtr& opsErrCtx，OpoConValCtx* pOpoConValCtx，OpoConRefCtx& pOpoConRefCtx) 在 Oracle.DataAccess.Client.ConnectionDispenser.Open(OpoConCtx opoConCtx) 在 Oracle.DataAccess.Client.OracleConnection.Open() 在 System.Data.EntityClient.EntityConnection.OpenStoreConnectionIf（布尔 openCondition, DbConnection storeConnectionToOpen, DbConnection originalConnection，字符串异常代码，字符串尝试操作， 布尔值& closeStoreConnectionOnFailure) 在 System.Data.EntityClient.EntityConnection.Open() 在 System.Data.Objects.ObjectContext.EnsureConnection() 在 System.Data.Objects.ObjectQuery1.GetResults(Nullable1 forMergeOption) 在 System.Data.Objects.ObjectQuery1.System.Collections.Generic.IEnumerable<T>.GetEnumerator() at System.Collections.Generic.List1..ctor（IEnumerable1 collection) at System.Linq.Enumerable.ToList[TSource](IEnumerable1 源） 在 ChangeOrder.Program.c_DisplayClass19.b_16(SF_CHANGE_ORDER_ITEM t) 在 C:\VS_apps\PMConsole\PMC 工具\ChangeOrderExecution\Program.cs：第 220 行 在 System.Threading.Tasks.Parallel.c_DisplayClass2d2.<ForEachWorker>b__23(Int32 i) at System.Threading.Tasks.Parallel.<>c__DisplayClassf1.b_c() 在 System.Threading.Tasks.Task.InnerInvoke() 在 System.Threading.Tasks.Task.InnerInvokeWithArg（任务子任务） 在 System.Threading.Tasks.Task.c_DisplayClass7.b_6（对象 ) 在 System.Threading.Tasks.Task.ExecuteSelfReplicating（任务根） 在 System.Threading.Tasks.Task.Execute() 在 System.Threading.Tasks.Task.ExecutionContextCallback（对象 obj） 在 System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback 回调, 对象状态, Boolean 忽略SyncCtx） 在 System.Threading.Tasks.Task.ExecuteWithThreadLocal（任务和 currentTaskSlot） 在 System.Threading.Tasks.Task.ExecuteEntry（布尔 bPreventDoubleExecution） 在 System.Threading.Tasks.ThreadPoolTask​​Scheduler.TryExecuteTaskInline（任务 任务，布尔任务之前已排队） 在 System.Threading.Tasks.TaskScheduler.TryRunInline（任务任务，布尔 taskWasPreviousQueued，对象 threadStatics） 在 System.Threading.Tasks.Task.InternalRunSynchronously（TaskScheduler 调度器） 在 System.Threading.Tasks.Task.RunSynchronously（TaskScheduler 调度程序） 在 System.Threading.Tasks.Parallel.ForWorker[TLocal](Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Action1 body, Action2 bodyWithState, Func4 bodyWithLocal, Func1 本地初始化，动作1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IList1 列表，ParallelOptions 并行选项，Action1 body, Action2 bodyWithState, 动作3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal, Func5 bodyWithEverything, Func1 localInit, 行动1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IEnumerable1 来源，ParallelOptions 并行选项，Action1 body, Action2 bodyWithState, 动作3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal, Func5 bodyWithEverything, Func1 localInit, Action1 localFinally) at System.Threading.Tasks.Parallel.ForEach[TSource](IEnumerable1 来源， 动作`1 正文） 在 C:\VS_apps\PMConsole\PMC 中的 ChangeOrder.Program.PerformChangeOrder(SF_CHANGE_ORDER co, SF_CLIENT_PROJECT cp, SFEntitiesQA ctx) 工具\ChangeOrderExecution\Program.cs：第 216 行 在 C:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program.cs:line 中的 ChangeOrder.Program.Main(String[] args) 1373 在 System.AppDomain._nExecuteAssembly（RuntimeAssembly 程序集，字符串 [] 参数） 在 System.AppDomain.ExecuteAssembly（字符串 assemblyFile，证据 assemblySecurity，String [] args） 在 Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly() 在 System.Threading.ThreadHelper.ThreadStart_Context（对象状态） 在 System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback 回调, 对象状态, Boolean 忽略SyncCtx） 在 System.Threading.ExecutionContext.Run（ExecutionContext executionContext，ContextCallback 回调，对象状态） 在 System.Threading.ThreadHelper.ThreadStart() InnerException：

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

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

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

Answer 1

为什么不单独进行连接而不是单独获取每个连接？由于这看起来像是在访问数据库，因此您的 LINQ 提供程序应该只编写查询并为您提供您正在寻找的行。试试这个：

List<SF_CO_LINE_ITEM> gufs;
var query = from coi in ctx.SF_CO_ITEM
            where coi.CO == co.ID
            join coli in ctx.SF_CO_LINE_ITEM      
                on coi.ID == coli.CO_ITEM
            select coli;
// Confirm what the query looks like by calling 'query.ToString()'
gufs = query.ToList();

我通常会将查询与实际的枚举/实现分开，这样我就可以验证查询是否符合我的要求。如果这是 SF_CO_ITEM 和 SF_CO_LINE_ITEM 之间的 1:M 关系，那么您应该通过将联接更改为：

join coli in ctx.SF_CO_LINE_ITEM      
    on coi.ID == coli.CO_ITEM into tcoli
from tc in tcoli
select tc

至于为什么会出现此异常，可能与尝试从不同线程访问上下文有关。根据MSDN article on Parallel loops：

隐藏循环体依赖

循环依赖的错误分析是软件的常见来源 缺陷。注意所有并行循环体不包含隐藏 依赖关系。这是一个很容易犯的错误。

尝试共享类的实例的情况，例如 Random 或 跨并行迭代的 DbConnection 不是线程安全的 一个微妙依赖的例子。

因此，您唯一的选择是保持顺序而不是并行，或者将原始查询更改为连接，以便您在第一时间获得正确的数据。

希望有帮助！