しばらくしてから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); 
    }