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

Question

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

我从 Parallel.Foreach SQL querying sometimes results in Connection 开始.我不想使用已接受的答案，因为它放弃了 Parallel.ForEach而只使用 Task.Run (或 Task Factory.StartNew 对于 .Net 4.0)。同时，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):

This delegate can be invoked concurrently by multiple tasks

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

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

正确的方法是使用 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 绑定(bind)的工作(比如进行 SQL 查询)不是一个好主意，使用 async\await 会做得更好。但这超出了这个问题的范围。