Parallel.ForEach在一段時間後會凍結

Question

我想使用Parallel.ForEach添加1000多個任務。以下代碼用於發送電子郵件通知。問題是，它只適用於約150〜200通知&我收到電子郵件，但在此之後代碼被凍結&沒有收到電子郵件。

有人可以請指導我在正確的方向。

var exceptions = new ConcurrentQueue<Exception>(); 

try 
{ 
    List<ParallelWorker_EmailNotification> workers = new List<ParallelWorker_EmailNotification>(); 

    foreach (Email mail in listEmails) 
    { 
     workers.Add(new ParallelWorker_EmailNotification(mail)); 
    } 

    Parallel.ForEach(workers, async worker => 
    { 
     try 
     { 
      await worker.SendNotification(); 
     } 
     catch (Exception ex) 
     { 
      exceptions.Enqueue(ex); 
     } 
    }); 
} 
catch (Exception ex) 
{ 
    exceptions.Enqueue(ex); 
} 

Answer 1

Parallel.ForEach不傳入異步函數工作的async worker =>方法簽名是async void而且有可能你的問題的根源。 Parallel.ForEach我們解鎖，因爲它認爲工作已完成，但工作仍在後臺處理，這就是爲什麼你看不到項目處理。

最簡單的解決方案（如果SendNotification是一個合適的異步函數）只是選擇項目並將所有任務都放到一個IEnumerable中並等待它們。

var exceptions = new ConcurrentQueue<Exception>(); 

    try 
    { 
     var tasks = listEmails.Select(mail => new ParallelWorker_EmailNotification(mail)) 
           .Select(async worker => 
      { 
       try 
       { 
        await worker.SendNotification(); 
       } 
       catch (Exception ex) 
       { 
        exceptions.Enqueue(ex); 
       } 
      }); 
     await Task.WhenAll(tasks); 
    } 
    catch (Exception ex) 
    { 
     exceptions.Enqueue(ex); 
    } 

如果SendNotification是一個函數，需要一些時間它產生控制之前返回給調用者的最佳解決方案是使用用於TPL Dataflow的處理。

var exceptions = new ConcurrentQueue<Exception>(); 

    try 
    { 
     var actionBlock = new ActionBlock<ParallelWorker_EmailNotification>(async worker => 
      { 
       try 
       { 
        await worker.SendNotification(); 
       } 
       catch (Exception ex) 
       { 
        exceptions.Enqueue(ex); 
       } 
      }, new ExecutionDataflowBlockOptions 
        { 
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded       
        } 
      ); 

     foreach (Email mail in listEmails) 
     { 
      actionBlock.Post(new ParallelWorker_EmailNotification(mail)); 
     } 
     actionBlock.Complete(); 
     actionBlock.Completion.Wait(); 
    } 
    catch (Exception ex) 
    { 
     exceptions.Enqueue(ex); 
    }