代码之家 › 专栏 › 技术社区 › Ahmad

使用锁对任务进行排队是一个好的做法吗?

locking c#

Ahmad · 技术社区 · 7 年前

我需要将一些可以在任何时候到达的请求放入队列中,使每个任务只在前一个任务结束时才开始。问题是,为此使用锁定是一个好主意吗?它有什么不好的影响吗?我所期望的排队行为会由此产生吗?

更具体地说,请考虑以下代码:

private int MyTask() {
   ...
}

private object someLock = new object();

public Task<int> DoMyTask() {
    return Task.Run(() =>
    {
        lock (someLock)
        {
            return MyTask();
        }
    });
}

public void CallMyTask() {
    var result = await DoMyTask();
}

请注意 CallMyTask() 将在任何时候调用,可能同时调用。

0 回复 | 直到 7 年前

Stephen Cleary 7 年前

为此使用锁定是个好主意吗?它有什么不好的影响吗?我所期望的排队行为会由此产生吗?

从工作在队列中开始 直到工作完成。因此,如果您的代码排队等待1000个请求,它将调用 Task.Run 1000次,并有可能耗尽该数量的线程池线程,每个线程除了等待锁之外什么都不做。

而且,锁并不是严格的FIFO。他们基本上都是先进先出。这是因为严格的FIFO锁会导致其他问题,比如锁护送; the links in this issue have some great discussion about lock "fairness" (i.e., FIFO behavior) .

所以,我推荐一个实际的队列。你可以用 ActionBlock<T> 从 TPL Dataflow 充当真正的队列。既然你的要求结果 ,您可以使用 TaskCompletionSource<T> 使排队代码能够得到结果。 任务完成源<T> 是一个“异步信号”——在本例中,我们使用它来通知调用代码请求已通过队列并被执行。

private ActionBlock<TaskCompletionSource<int>> queue =
    new ActionBlock<TaskCompletionSource<int>>(tcs =>
    {
      try { tcs.TrySetResult(MyTask()); }
      catch (Exception ex) { tcs.TrySetException(ex); }
    });

每次我们发送 任务完成源<T> 为了这个 queue ,它将运行 MyTask() 并捕获结果(无论是成功还是异常),并将这些结果传递给 任务完成源<T> .

我们可以这样使用它:

public Task<int> DoMyTask() {
  var tcs = new TaskCompletionSource<int>();
  queue.Post(tcs);
  return tcs.Task;
}

public void CallMyTask() {
  var result = await DoMyTask();
}

Jay 7 年前

我认为锁定是实现这一点的唯一方法,但是.NET框架应该能够为您做到这一点,如果您使用 blocking collection 还有一个 concurrent queue . Blocking集合为您提供了一个线程安全的producer/consumer模式的实现。


class Program
{
    private static BlockingCollection<Task> m_BlockingCollection = new BlockingCollection<Task>(new ConcurrentQueue<Task>());
    private static int Counter;

    static async Task Main(string[] args)
    {
        Task.Run(ProcessQueue); //Don't await for this demo!
        Task.Run(AddStuffToQueue); //Don't await for this demo!

        Console.ReadLine();
        m_BlockingCollection.CompleteAdding();
        while (!m_BlockingCollection.IsAddingCompleted)
            Thread.Sleep(5);
    }

    private static void AddStuffToQueue()
    {
        while(true)
            m_BlockingCollection.Add(new Task(() => Console.WriteLine(Interlocked.Increment(ref Counter))));
    }

    private static async Task ProcessQueue()
    {
        while (!m_BlockingCollection.IsCompleted && m_BlockingCollection.TryTake(out Task task))
            ProcessTask(task);
    }

    private static void ProcessTask(Task task)
    {
        task.RunSynchronously();
    }
}

这也许不是一个完美的例子,但我相信你明白了。生产者/消费者包装了并发队列,因此任务以先进先出(FIFO)的形式执行。

对于blocking集合,可以有多个使用者,但是如果您希望一次处理一个使用者,那么使用一个使用者就足够了。

我希望这有帮助!