这个 C# Parallel.ForEach "localFinally" 方法线程安全吗？

Question

由于 MicroSoft 所述的好处，我正在使用 Parallel.ForEach 实现大量 SQL SELECT 语句（数以千计）（顺便说一句：我现在找不到它，但我我很确定我在某处读到 Parallel.ForEach 在执行所有迭代之前不会返回控制权。这是真的吗？）

我从Parallel.Foreach SQL querying sometimes results in Connection 开始。我不想使用已接受的答案，因为它放弃了Parallel.ForEach，而只使用了Task.Run（或.Net 4.0 的Task Factory.StartNew）。同时，O.P. 使用“锁定”来同步更新到DataTable 的列表，我理解这可能会降低Parallel.ForEach 的效率。

所以，使用文章How to: Write a Parallel.ForEach Loop with Thread-Local Variables，我写了下面的稻草人代码。目标是将threadLocalDataTable（每个选择语句一个）累积到所有查询完成后返回的dataTableList。它可以工作，但我想知道localFinally 方法是否真的是线程安全的（请参阅带有注释//<-- the localFinally method in question 的代码行。注意：SqlOperation 类实现了连接字符串、输出数据表名称和选择查询字符串

public static IList<DataTable> GetAllData(IEnumerable<SqlOperation> sqlList)
{
    IList<DataTable> dataTableList = new List<DataTable>();
    dataTableList.Clear();

    try
    {

        Parallel.ForEach<SqlOperation, DataTable>(sqlList, () => null, (s, loop, threadLocalDataTable) =>
        {

            DataTable dataTable = null;

            using (SqlCommand sqlCommand = new SqlCommand())
            {
                using (SqlConnection sqlConnection = new SqlConnection(s.ConnectionString))
                {
                    sqlConnection.Open();
                    sqlCommand.CommandType = CommandType.Text;
                    sqlCommand.Connection = sqlConnection;
                    sqlCommand.CommandText = s.SelectQuery;
                    SqlDataAdapter sqlDataAdapter = new SqlDataAdapter(sqlCommand);

                    dataTable = new DataTable
                    {
                        TableName = s.DataTableName
                    };

                    dataTable.Clear();
                    dataTable.Rows.Clear();
                    dataTable.Columns.Clear();

                    sqlDataAdapter.Fill(dataTable);
                    sqlDataAdapter.Dispose();
                    sqlConnection.Close();
                }
            }

            return dataTable;


        }, (threadLocalDataTable) => dataTableList.Add(threadLocalDataTable) //<-- the localFinally method in question
        );
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex.ToString(), "GetAllData error");
    }

    return dataTableList;
}

Answer 1

您提供的相关链接的文档清楚地说明了（关于 localFinally）：

这个委托可以被多个任务同时调用

所以不，它不是线程安全的。但这不是唯一的问题 - 您的整个实现不正确。

传递给每个循环迭代的“线程局部”变量是累加器。它累积在同一线程上运行的多次迭代获得的结果。您的实现完全忽略了该变量并始终返回一个数据表。这意味着，如果您的表比并行循环的线程数（默认为处理器的内核数）多，您实际上会丢失结果，因为您忽略了累加器。

正确的方法是使用List<DataTable> 作为累加器，而不是一个累加器。例如：

Parallel.ForEach<SqlOperation, List<DataTable>>(
    sqlList, 
    () => new List<DataTable>(), // initialize accumulator
    (s, loop, threadLocalDataTables) =>
    {
        DataTable result = ...;
        // add, that's thread safe
        threadLocalDataTables.Add(result);
        // return accumulator to next iteration
        return threadLocalDataTables;
    }, (threadLocalDataTables) =>
    {
        //<-- the localFinally method in question
        lock (dataTableList) // lock and merge results
           dataTableList.AddRange(threadLocalDataTables);
    });

也就是说 - 使用 Parallel.ForEach 来加速 IO 密集型工作（例如进行 SQL 查询）并不是一个好主意，使用 async\await 会做得更好。但这超出了这个问题的范围。