Task.Run() 在高负载下随机/不规则地安排任务

Question

由于我正在处理大量电子邮件，我使用Task 方法来处理这些电子邮件asynchronously，而不会影响主要工作。 （基本上，发送电子邮件功能应该在与主线程不同的线程上工作。）假设您在 Windows Service 中每 30 秒处理超过 1K 的电子邮件。

我面临的问题是——很多时候Task 方法没有被执行，它完全是随机的。从技术上讲，它随机安排任务。有时我在 SendEmail 方法中收到电话，有时没有。两种方法我都试过了。

方法一

public void ProcessMails()
{
    Task.Run(() => SendEmail(emailModel));
}

方法二

public async void ProcessMails()
{
    // here the SendEmail method is awaitable, but I have not used 'await' because 
    // I need non-blocking operation on main thread.
    SendEmail(emailModel));
}

谁能告诉我可能是什么问题，或者我在这里遗漏了什么？

Answer 1

如前所述，您似乎用尽了资源来安排最终发送电子邮件的任务。现在，示例代码试图强制提供所有需要立即安排的工作。

另一个答案提供了使用阻塞集合的建议，但我认为有一种更简洁的方法。这个示例至少应该给你正确的想法。

using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace ClassLibrary1
{

    public class MailHandler
    {
        private EmailLibrary emailLibrary = new EmailLibrary();
        private ExecutionDataflowBlockOptions options = new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = Environment.ProcessorCount };
        private ActionBlock<string> messageHandler;

        public MailHandler() => messageHandler = new ActionBlock<string>(msg => DelegateSendEmail(msg), options);

        public Task ProcessMail(string message) => messageHandler.SendAsync(message);

        private Task DelegateSendEmail(string message) => emailLibrary.SendEmail(message);
    }

    public class EmailLibrary
    {
        public Task SendEmail(string message) => Task.Delay(1000);
    }
}

Answer 2

鉴于发送电子邮件的频率相当高，您可能为调度程序安排了过多的Tasks。

在方法一中，调用Task.Run每次都会创建一个新任务，每个任务都需要调度在一个线程上。这样做很可能会耗尽线程池。

虽然方法 2 会减少 Task 饥饿，即使有未等待的 Task 调用（即发即弃），async 方法之后的继续完成仍需要在 Threadpool 上调度，这将对您的系统产生不利影响.

代替未等待的Tasks 或Task.Run，并且由于您是Windows 服务，我将改为使用专门用于发送电子邮件的长时间运行的后台线程。该线程可以独立于您的主要工作，并且可以通过队列将电子邮件安排到该线程。

如果单个邮件发送线程不足以跟上邮件的同步，您可以扩展EmailSender 线程的数量，但将其限制为合理的有限数量。

您还应该探索其他优化，这将再次提高您的电子邮件发件人的吞吐量，例如

• 电子邮件发件人能否与邮件服务器保持长期连接？
• 邮件服务器是否接受批量邮件？

这是一个使用 BlockingCollection 的示例，并带有您的电子邮件模型的支持 ConcurrentQueue。

• 创建一个在生产者“PrimaryWork”线程和“EmailConsumer”线程之间共享的队列（显然，如果您有 IoC 容器，最好在那里注册）
• 在主工作线程上排队邮件
• 消费者EmailSender 在阻塞收集队列上运行一个循环，直到调用CompleteAdding
• 我使用 TaskCompletionSource 提供了一个任务，该任务将在所有消息发送完毕后完成，即可以优雅地退出而不会丢失仍在队列中的电子邮件。

public class PrimaryWork
{
    private readonly BlockingCollection<EmailModel> _enqueuer;

    public PrimaryWork(BlockingCollection<EmailModel> enqueuer)
    {
        _enqueuer = enqueuer;
    }

    public void DoWork()
    {
        // ... do your work
        for (var i = 0; i < 100; i++)
        {
          EnqueueEmail(new EmailModel { 
            To = $"recipient{i}@foo.com", 
            Message = $"Message {i}" });
        }
    }

    // i.e. Queue work for the email sender
    private void EnqueueEmail(EmailModel message)
    {
        _enqueuer.Add(message);
    }
}

public class EmailSender
{
    private readonly BlockingCollection<EmailModel> _mailQueue;
    private readonly TaskCompletionSource<string> _tcsIsCompleted 
        = new TaskCompletionSource<string>();

    public EmailSender(BlockingCollection<EmailModel> mailQueue)
    {
        _mailQueue = mailQueue;
    }

    public void Start()
    {
        Task.Run(() =>
        {
            try
            {
                while (!_mailQueue.IsCompleted)
                {
                    var nextMessage = _mailQueue.Take();
                    SendEmail(nextMessage).Wait();
                }
                _tcsIsCompleted.SetResult("ok");
            }
            catch (Exception)
            {
                _tcsIsCompleted.SetResult("fail");
            }
        });
    }

    public async Task Stop()
    {
        _mailQueue.CompleteAdding();
        await _tcsIsCompleted.Task;
    }

    private async Task SendEmail(EmailModel message)
    {
        // IO bound work to the email server goes here ...
    }
}

引导和启动上述生产者/消费者类的示例：

public static async Task Main(string[] args)
{
    var theQueue = new BlockingCollection<EmailModel>(new ConcurrentQueue<EmailModel>());
    var primaryWork = new PrimaryWork(theQueue);
    var mailSender = new EmailSender(theQueue);
    mailSender.Start();
    primaryWork.DoWork();
    // Wait for all mails to be sent 
    await mailSender.Stop();
}

我已经在Bitbucket here上放了一个完整的示例

其他说明

• 阻塞集合（和支持ConcurrentQueue）是线程安全的，因此您可以同时使用多个生产者和消费者线程。
• 如上所述，鼓励批处理，异步并行是可能的（由于每个邮件发送者使用一个线程，Task.WaitAll(tasks) 将等待一批任务）。一个完全异步的发送者显然可以使用await Task.WhenAll(tasks)。
• 根据下面的 cmets，我相信您的系统的性质（即 Windows 服务，每分钟有 2k 条消息）保证至少有一个专用线程用于电子邮件发送，尽管电子邮件可能天生就受 IO 限制。