c# - 使用 FIFO 同步运行任务的好方法？

Question

目前，我的第一步是使用 .NET 中的 async/await 和任务，异步运行是多么容易，这让我感到非常兴奋!但是，目前我必须通过 SerialPort 与设备通信。由于同时只能有一个连接，我只是编写了一些扩展方法来运行所有这些方法，这些方法来自不同的任务/线程，以先进先出的顺序同步运行:

public static class Extensions
{
    private readonly static object LockObject = new object();

    public static Task<TResult> RunAfter<TResult>(this Task<TResult> task, ConcurrentQueue<Task> others)
        => (Task<TResult>)task.RunAllSynchronously(others);

    public static Task RunAfter(this Task task, ConcurrentQueue<Task> others)
        => task.RunAllSynchronously(others);

    private static Task RunAllSynchronously(this Task task, ConcurrentQueue<Task> others)
    {
        if (others == null) throw new ArgumentNullException("The value of " + nameof(others) + " is null!");
        lock (LockObject)
        {
            others.Enqueue(task);
            Task currentTask;
            while (others.TryDequeue(out currentTask))
            {
                currentTask.RunSynchronously();
                if (currentTask == task) break;
            }
        }
        return task;
    }
}

这种方法似乎是一种好方法，还是应该区别对待这种情况？

Answer 1

为什么要同步运行它们？

您应该异步运行任务并使用 async和 await一个一个地执行它们:

 Task currentTask;
 while (others.TryDequeue(out currentTask))
 {
      await currentTask;
      if (currentTask == task) break;
 }

另一方面，查看您的代码，我找不到使用 lock 的理由(线程同步)。 您根据某些共享资源同步线程(某些对象可能会或可能不会被多个线程读取/修改)。您的方法可以修改为:

private static async Task RunAllAsync(this Task task, ConcurrentQueue<Task> others)
{
    // Design by contract rocks ;)
    // See:  https://msdn.microsoft.com/en-us/library/dd264808(v=vs.110).aspx
    Contracts.Requires(task != null);
    Contracts.Requires(others != null);

    others.Enqueue(task);

    // See how I've improved your loop. Since ConcurrentQueue.TryDequeue
    // will return false if other thread has called it already, your loop
    // should try to dequeue again until it returns true, and it should
    // break if dequeued task is the task against which the extension method
    // was called or the concurrent queue has no more items, to prevent a 
    // possible infinite loop
    do
    { 
       Task currentTask;
       if(others.TryDequeue(out currentTask))
          await currentTask;

    }
    while (currentTask == task || others.Count > 0);

    return task;
}

更新

OP 说:

I have possibly forgotten to say, that the ConcurrentQueue is the resource that should be shared among the threads. I.e. Task.RunAllSynchronously() is called on every new Task (access to SerialPort) and this call could be come from a different thread. Also, I cannot ensure that RunAllSynchronously() is just called, when all currently running (or queued) tasks are finished (I could, but therefore i had to use something like lock outside the extension method, which is not really that nice of having an extension method.

这就是您使用 ConcurrentQueue<T> 的原因.线程安全是在内部管理的。如果您调用 ConcurrentQueue<T>.TryDequeue并且不止一个线程同时调用它，只有一个会获胜，其他线程会收到 false作为返回值和 out不会分配参数。看什么MSDN says for this :

ConcurrentQueue handles all synchronization internally. If two threads call TryDequeue at precisely the same moment, neither operation is blocked. When a conflict is detected between two threads, one thread has to try again to retrieve the next element, and the synchronization is handled internally.

TryDequeue tries to remove an element from the queue. If the method is successful, the item is removed and the method returns true; otherwise, it returns false. That happens atomically with respect to other operations on the queue. If the queue was populated with code such as q.Enqueue("a"); q.Enqueue("b"); q.Enqueue("c"); and two threads concurrently try to dequeue an element, one thread will dequeue a and the other thread will dequeue b. Both calls to TryDequeue will return true, because they were both able to dequeue an element. If each thread goes back to dequeue an additional element, one of the threads will dequeue c and return true, whereas the other thread will find the queue empty and will return false.