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);
    }