.NET 中在单独（单个）线程上管理任务队列的最佳方式

Question

我知道这些年来异步编程发生了很多变化。让我在 34 岁的时候就这么生疏让我有些尴尬，但我指望 StackOverflow 能让我跟上进度。

我想要做的是在一个单独的线程上管理一个“工作”队列，但是一次只处理一个项目。我想在这个线程上发布工作，它不需要将任何东西传回给调用者。当然，我可以简单地启动一个新的Thread 对象，并让它在共享的Queue 对象上循环，使用睡眠、中断、等待句柄等。但我知道从那以后事情变得更好了。我们有BlockingCollection、Task、async/await，更不用说可能抽象出很多内容的 NuGet 包了。

我知道“什么是最好的……”问题通常不受欢迎，所以我将改写为“目前推荐的……”使用内置 .NET 机制完成此类事情的方式最好。但是，如果第三方 NuGet 包可以简化很多事情，那也一样。

我考虑了一个TaskScheduler 实例，它的最大并发数固定为 1，但现在似乎有一种不那么笨重的方法来做到这一点。

背景

具体来说，在这种情况下，我要做的是在 Web 请求期间排队 IP 地理定位任务。同一个 IP 可能会多次排队等待地理定位，但该任务将知道如何检测到这一点，如果已经解决，则提前跳过。但请求处理程序只是将这些() => LocateAddress(context.Request.UserHostAddress) 调用放入队列，并让LocateAddress 方法处理重复工作检测。我使用的地理定位 API 不喜欢被请求轰炸，这就是为什么我想一次将其限制为单个并发任务。但是，如果允许该方法通过简单的参数更改轻松扩展到更多并发任务，那就太好了。

Answer 1

要创建异步单度并行工作队列，您可以简单地创建一个 SemaphoreSlim，初始化为 1，然后在开始请求的工作之前获取该信号量的入队方法 await。

public class TaskQueue
{
    private SemaphoreSlim semaphore;
    public TaskQueue()
    {
        semaphore = new SemaphoreSlim(1);
    }

    public async Task<T> Enqueue<T>(Func<Task<T>> taskGenerator)
    {
        await semaphore.WaitAsync();
        try
        {
            return await taskGenerator();
        }
        finally
        {
            semaphore.Release();
        }
    }
    public async Task Enqueue(Func<Task> taskGenerator)
    {
        await semaphore.WaitAsync();
        try
        {
            await taskGenerator();
        }
        finally
        {
            semaphore.Release();
        }
    }
}

当然，要获得固定程度的并行度而不是一个，只需将信号量初始化为其他数字即可。

Answer 2

在我看来，您最好的选择是使用TPL Dataflow 的ActionBlock：

var actionBlock = new ActionBlock<string>(address =>
{
    if (!IsDuplicate(address))
    {
        LocateAddress(address);
    }
});

actionBlock.Post(context.Request.UserHostAddress);

TPL Dataflow 是健壮的、线程安全的、async-ready 和非常可配置的基于 actor 的框架（以 nuget 形式提供）

这是一个更复杂情况的简单示例。假设您想要：

• 启用并发（仅限于可用内核）。
• 限制队列大小（这样您就不会耗尽内存）。
• LocateAddress 和队列插入都是async。
• 一小时后取消所有内容。

var actionBlock = new ActionBlock<string>(async address =>
{
    if (!IsDuplicate(address))
    {
        await LocateAddressAsync(address);
    }
}, new ExecutionDataflowBlockOptions
{
    BoundedCapacity = 10000,
    MaxDegreeOfParallelism = Environment.ProcessorCount,
    CancellationToken = new CancellationTokenSource(TimeSpan.FromHours(1)).Token
});

await actionBlock.SendAsync(context.Request.UserHostAddress);

Answer 3

实际上你不需要在一个线程中运行任务，你需要它们串行运行（一个接一个），并且先进先出。 TPL 没有这方面的课程，但这是我的非常轻量级、非阻塞的测试实现。 https://github.com/Gentlee/SerialQueue

那里也有@Servy 实现，测试表明它比我的慢两倍，并且不能保证 FIFO。

例子：

private readonly SerialQueue queue = new SerialQueue();

async Task SomeAsyncMethod()
{
    var result = await queue.Enqueue(DoSomething);
}

Answer 4

使用BlockingCollection<Action> 创建一个生产者/消费者模式，其中包含一个消费者（一次只运行一件事，如您所愿）和一个或多个生产者。

首先在某处定义一个共享队列：

BlockingCollection<Action> queue = new BlockingCollection<Action>();

在您的消费者 Thread 或 Task 中，您可以从中获取：

//This will block until there's an item available
Action itemToRun = queue.Take()

然后从其他线程上的任意数量的生产者，简单地添加到队列中：

queue.Add(() => LocateAddress(context.Request.UserHostAddress));

Answer 5

我在这里发布了一个不同的解决方案。老实说，我不确定这是否是一个好的解决方案。

我习惯于使用 BlockingCollection 来实现生产者/消费者模式，并使用一个专用线程来消耗这些项目。如果总是有数据进来并且消费者线程不会坐在那里什么都不做，那很好。

我遇到了这样一种情况，其中一个应用程序想在不同的线程上发送电子邮件，但电子邮件的总数并没有那么大。 我最初的解决方案是拥有一个专用的消费者线程（由 Task.Run() 创建），但很多时候它只是坐在那里，什么都不做。

旧解决方案：

private readonly BlockingCollection<EmailData> _Emails =
    new BlockingCollection<EmailData>(new ConcurrentQueue<EmailData>());

// producer can add data here
public void Add(EmailData emailData)
{
    _Emails.Add(emailData);
}

public void Run()
{
    // create a consumer thread
    Task.Run(() => 
    {
        foreach (var emailData in _Emails.GetConsumingEnumerable())
        {
            SendEmail(emailData);
        }
    });
}

// sending email implementation
private void SendEmail(EmailData emailData)
{
    throw new NotImplementedException();
}

如您所见，如果没有足够的电子邮件要发送（这是我的情况），消费者线程将把大部分电子邮件都花在那里，什么也不做。

我将实现更改为：

// create an empty task
private Task _SendEmailTask = Task.Run(() => {});

// caller will dispatch the email to here
// continuewith will use a thread pool thread (different to
// _SendEmailTask thread) to send this email
private void Add(EmailData emailData)
{
    _SendEmailTask = _SendEmailTask.ContinueWith((t) =>
    {
        SendEmail(emailData);
    });
}

// actual implementation
private void SendEmail(EmailData emailData)
{
    throw new NotImplementedException();
}

它不再是生产者/消费者模式，但它不会有一个线程坐在那里什么都不做，而是每次发送电子邮件时，它都会使用线程池线程来做。

Answer 6

我的库，它可以：

1. 在队列列表中随机运行
2. 多队列
3. 优先运行
4. 重新排队
5. 事件所有队列完成
6. 取消运行或取消等待运行
7. 向 UI 线程调度事件

---

public interface IQueue
  {
    bool IsPrioritize { get; }
    bool ReQueue { get; }
    /// <summary>
    /// Dont use async
    /// </summary>
    /// <returns></returns>
    Task DoWork();
    bool CheckEquals(IQueue queue);
    void Cancel();
  }

  public delegate void QueueComplete<T>(T queue) where T : IQueue;
  public delegate void RunComplete();
  public class TaskQueue<T> where T : IQueue
  {
    readonly List<T> Queues = new List<T>();
    readonly List<T> Runnings = new List<T>();

    [Browsable(false), DefaultValue((string)null)]
    public Dispatcher Dispatcher { get; set; }
    public event RunComplete OnRunComplete;
    public event QueueComplete<T> OnQueueComplete;
    int _MaxRun = 1;
    public int MaxRun
    {
      get { return _MaxRun; }
      set
      {
        bool flag = value > _MaxRun;
        _MaxRun = value;
        if (flag && Queues.Count != 0) RunNewQueue();
      }
    }
    public int RunningCount
    {
      get { return Runnings.Count; }
    }
    public int QueueCount
    {
      get { return Queues.Count; }
    }

    public bool RunRandom { get; set; } = false;

    //need lock Queues first
    void StartQueue(T queue)
    {
      if (null != queue)
      {
        Queues.Remove(queue);
        lock (Runnings) Runnings.Add(queue);
        queue.DoWork().ContinueWith(ContinueTaskResult, queue);
      }
    }

    void RunNewQueue()
    {
      lock (Queues)//Prioritize
      {
        foreach (var q in Queues.Where(x => x.IsPrioritize)) StartQueue(q);
      }

      if (Runnings.Count >= MaxRun) return;//other
      else if (Queues.Count == 0)
      {
        if (Runnings.Count == 0 && OnRunComplete != null)
        {
          if (Dispatcher != null && !Dispatcher.CheckAccess()) Dispatcher.Invoke(OnRunComplete);
          else OnRunComplete.Invoke();//on completed
        }
        else return;
      }
      else
      {
        lock (Queues)
        {
          T queue;
          if (RunRandom) queue = Queues.OrderBy(x => Guid.NewGuid()).FirstOrDefault();
          else queue = Queues.FirstOrDefault();
          StartQueue(queue);
        }
        if (Queues.Count > 0 && Runnings.Count < MaxRun) RunNewQueue();
      }
    }

    void ContinueTaskResult(Task Result, object queue_obj) => QueueCompleted((T)queue_obj);

    void QueueCompleted(T queue)
    {
      lock (Runnings) Runnings.Remove(queue);
      if (queue.ReQueue) lock (Queues) Queues.Add(queue);
      if (OnQueueComplete != null)
      {
        if (Dispatcher != null && !Dispatcher.CheckAccess()) Dispatcher.Invoke(OnQueueComplete, queue);
        else OnQueueComplete.Invoke(queue);
      }
      RunNewQueue();
    }

    public void Add(T queue)
    {
      if (null == queue) throw new ArgumentNullException(nameof(queue));
      lock (Queues) Queues.Add(queue);
      RunNewQueue();
    }

    public void Cancel(T queue)
    {
      if (null == queue) throw new ArgumentNullException(nameof(queue));
      lock (Queues) Queues.RemoveAll(o => o.CheckEquals(queue));
      lock (Runnings) Runnings.ForEach(o => { if (o.CheckEquals(queue)) o.Cancel(); });
    }

    public void Reset(T queue)
    {
      if (null == queue) throw new ArgumentNullException(nameof(queue));
      Cancel(queue);
      Add(queue);
    }

    public void ShutDown()
    {
      MaxRun = 0;
      lock (Queues) Queues.Clear();
      lock (Runnings) Runnings.ForEach(o => o.Cancel());
    }
  }