并行执行存储过程。foreach .net答案

【问题标题】：Executing stored procedures in parallel.foreach .net并行执行存储过程。foreach .net
【发布时间】：2015-12-11 13:35:56
【问题描述】：

我需要运行 4 个存储过程并根据收到的数据创建查询。 q 容器中大约有 5k 个项目，它是存储过程的 20k 次执行。我使用 LINQ 连接到 DB 并执行它们，它与普通的 foreach 循环配合得很好，但有一个问题：代码大约需要一小时才能完成。这太长了，所以我尝试编写 Parrarel.ForEach 而不是普通的 ForEach 循环。几次迭代后代码崩溃 - 我猜 LINQ 连接无法与 Parrarel 连接。任何想法如何在多个线程中运行 LINQ 存储过程？

var dataCollector = new EpmDataCollector();
Parallel.ForEach(q, history =>
{                    
try
{  
        var queriesBefore = dataCollector.GetQueries().Count;
        var weight = dataCollector.CreateProjectQuery(history);//function executes stored procedure and creates queries from data received, then adds them to container (ConcurrentBag) in dataCollector
        dataCollector.CreateHoursQuery(history);//like above
        dataCollector.CreateCostQuery(history);//same
        dataCollector.CreateIncomeQuery(history);//same                    

        var log = ...
        Global.log.Info(log);

        //i++;
        Interlocked.Increment(ref i);
        if (i % 10 == 0)
        {
            //calculate and log estimation time
        } 

}
catch (Exception ex)
{
    //catch code
}
});

【问题讨论】：

是的；在 foreach 循环中创建新的 linq 上下文。
我也尝试锁定 dataCollector 但它似乎什么也没做。
@OndrejSvejdar 然后我在 SqlClient.TdsParser.TryRun 中得到了 AggregateException
聚合异常通常意味着“一个或多个线程出现问题”。聚合异常下的异常列表包含重要的东西——那有什么？如果你需要帮助，你必须尽可能描述性:)

标签： c# .net multithreading linq parallel.foreach

【解决方案1】：

System.Data.Linq.DataContext 类不是线程安全的。

参考：https://msdn.microsoft.com/en-us/library/system.data.linq.datacontext(v=vs.110).aspx

这种类型的任何公共静态成员都是线程安全的。不保证任何实例成员都是线程安全的。

这就是为什么你必须在ForEach 循环中创建DataContext 的新实例。

另外我更愿意研究专为处理数千次插入而设计的SqlBulkCopy (https://msdn.microsoft.com/en-us/library/system.data.sqlclient.sqlbulkcopy(v=vs.110).aspx)。

【讨论】：

我不需要插入数据。我执行存储过程并处理收到的数据

【解决方案2】：

将 EpmDataCollector 的创建移动到循环中，如下所示：

Parallel.ForEach(q, history =>
{                    
try
{  
        var dataCollector = new EpmDataCollector();
        var queriesBefore = dataCollector.GetQueries().Count;
        var weight = dataCollector.CreateProjectQuery(history);//function executes stored procedure and creates queries from data received, then adds them to container (ConcurrentBag) in dataCollector
        dataCollector.CreateHoursQuery(history);//like above
        dataCollector.CreateCostQuery(history);//same
        dataCollector.CreateIncomeQuery(history);//same                    

        var log = ...
        Global.log.Info(log);

        //i++;
        Interlocked.Increment(ref i);
        if (i % 10 == 0)
        {
            //calculate and log estimation time
        } 

}
catch (Exception ex)
{
    //catch code
}
});

【讨论】：