在Parallel.ForEach和Task.Factory.StartNew中記錄每個項目的異常

Question

我想在列表中使用Parallel.ForEach，並嘗試使用列表中的每個項目嘗試進行數據庫調用。我正在嘗試記錄每個項目是否有錯誤。只是想在這裏與專家覈對如果我正在做正確的方式。對於這個例子，我使用文件訪問而不是數據庫訪問來模擬I/O。

static ConcurrentQueue<IdAndErrorMessage> queue = new ConcurrentQueue<IdAndErrorMessage>(); 
    private static void RunParallelForEach() 
    { 
     List<int> list = Enumerable.Range(1, 5).ToList<int>(); 
     Console.WriteLine("Start...."); 
     Stopwatch stopWatch = new Stopwatch(); 
     stopWatch.Start(); 
     Parallel.ForEach(list, (tempId) => 
     { 
     string errorMessage = string.Empty; 
     try 
     { 
      ComputeBoundOperationTest(tempId); 
      try 
      { 
       Task[] task = new Task[1] 
       { 
       Task.Factory.StartNew(() => this.contentFactory.ContentFileUpdate(content, fileId)) 
       }; 
      } 
      catch (Exception ex) 
      { 
       this.tableContentFileConversionInfoQueue.Enqueue(new ContentFileConversionInfo(fileId, ex.ToString())); 
      } 
     } 
     catch (Exception ex) 
     { 
      errorMessage = ex.ToString(); 
     } 
     if (queue.SingleOrDefault((IdAndErrorMessageObj) => IdAndErrorMessageObj.Id == tempId) == null) 
     { 
      queue.Enqueue(new IdAndErrorMessage(tempId, errorMessage)); 
     } 
    } 
    ); 
    Console.WriteLine("Stop...."); 
    Console.WriteLine("Total milliseconds :- " + stopWatch.ElapsedMilliseconds.ToString()); 
} 

下面是輔助方法： -

private static byte[] FileAccess(int id) 
{ 
    if (id == 5) 
    { 
     throw new ApplicationException("This is some file access exception"); 
    } 
    return File.ReadAllBytes(Directory.GetFiles(Environment.SystemDirectory).First()); 
      //return File.ReadAllBytes("Files/" + fileName + ".docx"); 
} 

private static void ComputeBoundOperationTest(int tempId) 
{ 
    //Console.WriteLine("Compute-bound operation started for :- " + tempId.ToString()); 
    if (tempId == 4) 
    { 
     throw new ApplicationException("Error thrown for id = 4 from compute-bound operation"); 
    } 
    Thread.Sleep(20); 
} 

private static void EnumerateQueue(ConcurrentQueue<IdAndErrorMessage> queue) 
{ 
    Console.WriteLine("Enumerating the queue items :- "); 
    foreach (var item in queue) 
    { 
     Console.WriteLine(item.Id.ToString() + (!string.IsNullOrWhiteSpace(item.ErrorMessage) ? item.ErrorMessage : "No error")); 
    } 
} 

Answer 1

沒有理由這樣做：

/*Below task is I/O bound - so do this Async.*/ 
Task[] task = new Task[1] 
{ 
    Task.Factory.StartNew(() => FileAccess(tempId)) 
}; 
Task.WaitAll(task); 

通過在一個單獨的任務調度一下，然後馬上就可以等待，你只是綁定更多的線程。你最好把這個作爲：

/*Below task is I/O bound - but just call it.*/ 
FileAccess(tempId); 

話雖這麼說，因爲你正在做的每項目一個記錄值（異常或成功），你可能要考慮寫入的方法該然後把整個事情稱爲PLINQ查詢。

例如，如果你寫的這個成處理的try/catch（不帶螺紋），並返回「登錄串」的方法，即：

string ProcessItem(int id) { // ... 

你可以寫整個操作：

var results = theIDs.AsParallel().Select(id => ProcessItem(id)); 

Answer 2

Parallel.ForEach同時運行動作/ func達到一定數量的同時發生的事件。如果這些迭代中的每一個所做的都不是固有的彼此獨立的，那麼你沒有獲得任何性能收益。而且，可能會通過引入昂貴的上下文切換和爭用來降低性能。你說你想做一個「數據庫調用」並用文件操作來模擬它。如果每次迭代使用相同的資源（例如數據庫表中的同一行;或者嘗試寫入同一位置的同一文件），那麼它們並不會真正並行運行。一次只會有一個正在運行，其他人將只是「等待」以獲取資源 - 不必要地讓代碼變得複雜。

你還沒有詳細說明你想要爲每個迭代做什麼;但是當我遇到類似於其他程序員的情況時，他們幾乎總是沒有真正並行的做事，他們只是經歷了並且用Parallel.ForEach代替foreach s，希望神奇地獲得性能或神奇地使用多重-CPU/Core處理器。

Answer 3

您可能想要從線程代碼中刪除Console.WriteLine。原因是每個Windows應用程序只能有一個控制檯。所以如果兩個或多個線程要並行寫入控制檯，就必須等待。

取代您的自定義錯誤隊列，您可能希望看到.NET 4's Aggregate Exception並捕獲並相應地處理異常。 InnerExceptions propery將爲您提供必要的例外清單。更多here

而且一個通用代碼審查評論，請不要使用像4這樣的神奇數字在if (tempId == 4)而是定義一些可以告訴4代表什麼的const。例如if (tempId == Error.FileMissing)