用Parallel.ForEach实现Azure Parallel.ForEach线程安全

Question

我有一些基本的Azure表，我一直在不断地查询：

var query = new TableQuery<DynamicTableEntity>()
  .Where(TableQuery.GenerateFilterCondition("PartitionKey",
    QueryComparisons.Equal, myPartitionKey));

foreach (DynamicTableEntity entity in myTable.ExecuteQuery(query)) {
  // Process entity here.
}

为了加快速度，我将其并行化如下：

Parallel.ForEach(myTable.ExecuteQuery(query), (entity, loopState) => {
  // Process entity here in a thread-safe manner.

  // Edited to add: Details of the loop body below:

  // This is the essence of the fixed loop body:
  lock (myLock) {
    DataRow myRow = myDataTable.NewRow();
    // [Add entity data to myRow.]
    myDataTable.Rows.Add(myRow);
  }

  // Old code (apparently not thread-safe, though NewRow() is supposed to create
  // a DataRow based on the table's schema without changing the table state):
  /*
    DataRow myRow = myDataTable.NewRow();
    lock (myLock) {
      // [Add entity data to myRow.]
      myDataTable.Rows.Add(myRow);
    }
  */
});

这会产生显著的加速，但结果在运行之间往往略有不同(即，一些实体偶尔会有所不同，尽管返回的实体数量完全相同)。

通过这个和一些网络搜索，我得出结论，上面的枚举器并不总是线程安全的.文档似乎表明，只有在表对象是公共静态的情况下，线程安全才能得到保证，但这对我并没有什么影响。

有人能建议怎么解决这个问题吗？是否有并行化Azure表查询的标准模式？

Answer 1

您的评论是正确的: DataTable不适合于涉及突变的并发操作，并且是重复条目的来源。为行修改操作锁定DataTable对象将解决以下问题：

 lock (myTable)
 {
    DataRow myRow = myTable.NewRow();
    myRow.SetField<int>("c1", (int)value);
    myTable.Rows.Add(myRow);
 }

将NewRow()放置在锁之外会间歇性地导致表中重复的行条目，或者在NewRow()行中发生“System.Data.dll中出现'System.ArgumentException‘类型的未处理异常”异常。有关并发DataTable使用的其他详细信息和替代方案，请参阅Thread safety for DataTable。

若要再现错误条件，请使用以下代码。一些运行将是干净的，一些将包含重复的条目，还有一些将遇到异常。

   class Program
   {
      static DataTable myTable = GetTable();
      static ManualResetEvent waitHandle = new ManualResetEvent(false);

      static void Main(string[] args)
      {
         const int threadCount = 10;
         List<Thread> threads = new List<System.Threading.Thread>();
         for (int i = 0; i < threadCount; ++i) 
         {
            threads.Add(new Thread(new ParameterizedThreadStart(AddRowThread)));
            threads[i].Start(i);
         }
         waitHandle.Set(); // Release all the threads at once
         for (int i = 0; i < threadCount; ++i) 
         {
            threads[i].Join();
         }

         // Print results once threads return
         for (int i = 0; i < myTable.Rows.Count; ++i)
         {
            Console.WriteLine(myTable.Rows[i].Field<int>(0));
         }
         Console.WriteLine("---Processing Complete---");
         Console.ReadKey();
      }

      static void AddRowThread(object value)
      {
         waitHandle.WaitOne();
         DataRow myRow = myTable.NewRow(); // THIS RESULTS IN INTERMITTENT ERRORS
         lock (myTable)
         {
            //DataRow myRow = myTable.NewRow(); // MOVE NewRow() CALL HERE TO RESOLVE ISSUE
            myRow.SetField<int>("c1", (int)value);
            myTable.Rows.Add(myRow);
         }
      }

      static DataTable GetTable()
      {
         // Here we create a DataTable with four columns.
         DataTable table = new DataTable();
         table.Columns.Add("c1", typeof(int));       
         return table;
      }
   }