并行发送批量电子邮件的最佳方法

Question

我是 TPL（任务并行库）的新手，我很难将我的进程配置为并行运行任务。

我正在开发一个发送大量电子邮件的应用程序（比如每分钟数千封，这就是我的想法），但是当我看到处理器性能时，它并不好：我很确定会有很多开销因为我没有正确使用任务库。

这是我的代码：

public async void MainProcess()
{
    var batches = emailsToProcess.Batch(CONST_BATCHES_SIZE);
    
    foreach (var batch in batches.AsParallel()
        .WithDegreeOfParallelism(Environment.ProcessorCount))
    {
         await Task.WhenAll(from emailToProcess in batch 
                    select ProcessSingleEmail(emailToProcess));
        _emailsToProcessRepository.MarkBatchAsProcessed(batch);
    }
}

private async Task ProcessSingleEmail(EmailToProcess emailToProcess)
{
    try
    {
        MailMessage mail = GetMail(emailToProcess); //static light method
        await _smtpClient.SendAsync(sendGridMail);
        emailToProcess.Processed = true;
    }
    catch (Exception e)
    {
        _logger.Error(ErrorHelper.GetExceptionMessage(e, 
                    string.Format("Error sending Email ID #{0} : ", 
                    emailToProcess.Id)), e);
    }
}

（我知道它可能看起来很糟糕：请随意烤我☺）

我需要它以这种方式运行：我需要批量处理多个记录（顺便说一句，我正在使用允许我使用“批处理”方法的库），因为我需要标记一个批处理进程完成发送时在数据库中处理的记录数。

这个过程实际上正在做我想做的事：除了慢得要命。正如您在 perfmon 中看到的那样，处理器并没有以非常高的容量工作：

最好的方法是什么？有什么建议吗？

编辑：我意识到我所拥有的是一个开销问题。 有什么工具或简单的方法可以检测和纠正它们吗？

Answer 1

您所做的不是 CPU 限制，而是 I/O 限制，因此如果处理器可能会影响您的性能，请使用将并发任务的数量限制为数量。尝试并行启动更多任务。

例如，下面的代码将异步处理所有电子邮件，但并行处理限制为 100 封电子邮件。它使用ForEachAsync 扩展方法进行处理，该方法允许使用参数限制并行度，因此我会尝试尝试使该参数更大。

如果可能，您可能还希望使 MarkBatchAsProcessed 方法异步，因为这也可能会限制性能。

public static class Extensions
{
    public static async Task ExecuteInPartition<T>(IEnumerator<T> partition, Func<T, Task> body)
    {
        using (partition)
            while (partition.MoveNext())
                await body(partition.Current);
    }

    public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
    {
        return Task.WhenAll(
            from partition in Partitioner.Create(source).GetPartitions(dop)
            select ExecuteInPartition(partition, body));
    }
}

public Task MainProcess()
{
    // Process 100 emails at a time
    return emailsToProcess.ForEachAsync(100, async (m) =>
    {
        await ProcessSingleEmail(m);                
    });

    _emailsToProcessRepository.MarkBatchAsProcessed(emailsToProcess);
}

您还应该避免使用 void 返回异步方法，它们不会传播异常并且无法组合或等待，并且它们主要用于事件处理程序，因此我将 MainProcess 更改为返回 Task。

更新

上面代码中的数字 100 表示任何时候最多有 100 个并发任务，所以它更像是一个滑动窗口而不是一个批处理。如果您想批量处理电子邮件，可以执行以下操作（假设批次具有 Count 属性：

public async Task MainProcess()
{
    var batches = emailsToProcess.Batch(CONST_BATCHES_SIZE);

    foreach (var batch in batches)
    {
         return batch.ForEachAsync(batch.Count, async (m) =>
         {
             await ProcessSingleEmail(m);                
         });

       _emailsToProcessRepository.MarkBatchAsProcessed(batch);             
    }
}