.net - 用于单线程异步 async-await 风格编程的信号量

Question

信号量是一种多线程锁定机制，可确保在给定资源上仅运行有限数量的线程。互斥体是一种特殊情况，其中有限数量为 1。

异步编程与(有时与)多线程编程有很多共同点，尽管它本身并不是多线程的。

以下代码创建了 10 个任务，只需等待一秒钟并记录它们的开始和结束。

所有这些都只在一个线程上执行(我假设适当的同步上下文维护已经到位，例如 WPF 中的情况)。

因此，即使只有一个线程，我们也有“并行”任务，并且在某些用例中可能希望将资源访问权限限制为这些任务中的几个或一个。 (例如，限制并行网络请求。)

似乎需要一个“异步信号量”——一个不是锁定线程而是锁定异步延续的概念。

我已经实现了这样一个信号量来检查它是否真的有意义并阐明我的意思。

我的问题是:这个东西是否已经可用，最好在 .NET 框架本身中？我找不到任何东西，尽管在我看来它应该存在。

所以这是代码( LINQPad share here ):

    async void Main()
    {
        // Necessary in LINQPad to ensure a single thread.
        // Other environments such as WPF do this for you.
        SynchronizationContext.SetSynchronizationContext(
            new DispatcherSynchronizationContext());

        var tasks = Enumerable.Range(1, 10).Select(SampleWork).ToArray();

        await Task.WhenAll(tasks);

        "All done.".Dump();
    }

    AsyncSemaphore commonSemaphore = new AsyncSemaphore(4);

    async Task SampleWork(Int32 i)
    {
        using (await commonSemaphore.Acquire())
        {
            $"Beginning work #{i} {Thread.CurrentThread.ManagedThreadId}".Dump();

            await Task.Delay(TimeSpan.FromSeconds(1));

            $"Finished work #{i} {Thread.CurrentThread.ManagedThreadId}".Dump();
        }
    }

    public class AsyncSemaphore
    {
        Int32 maxTasks;
        Int32 currentTasks;

        ReleasingDisposable release;

        Queue<TaskCompletionSource<Object>> continuations
            = new Queue<TaskCompletionSource<Object>>();

        public AsyncSemaphore(Int32 maxTasks = 1)
        {
            this.maxTasks = maxTasks;
            release = new ReleasingDisposable(this);
        }

        public async Task<IDisposable> Acquire()
        {
            ++currentTasks;

            if (currentTasks > maxTasks)
            {
                var tcs = new TaskCompletionSource<Object>();

                continuations.Enqueue(tcs);

                await tcs.Task;
            }

            return release;
        }

        void Release()
        {
            --currentTasks;

            if (continuations.Count > 0)
            {
                var tcs = continuations.Dequeue();

                tcs.SetResult(null);
            }
        }

        class ReleasingDisposable : IDisposable
        {
            AsyncSemaphore self;

            public ReleasingDisposable(AsyncSemaphore self) => this.self = self;

            public void Dispose() => self.Release();
        }
    }

我得到这个输出:

Beginning work #1 1
Beginning work #2 1
Beginning work #3 1
Beginning work #4 1
Finished work #4 1
Finished work #3 1
Finished work #2 1
Finished work #1 1
Beginning work #5 1
Beginning work #6 1
Beginning work #7 1
Beginning work #8 1
Finished work #5 1
Beginning work #9 1
Finished work #8 1
Finished work #7 1
Finished work #6 1
Beginning work #10 1
Finished work #9 1
Finished work #10 1
All done.

所以确实，我最多运行 4 个任务，并且都在同一个线程上运行。

Answer 1

So even though there's only one thread we have "parallel" tasks

我通常更喜欢“并发”一词，以免与 Parallel 混淆。/并行 LINQ。

My question would be: Is this thing available already, ideally in the .NET framework itself?

是的。 SemaphoreSlim是一个可以同步使用的信号量 或 异步。

我也有 full suite of asynchronous coordination primitives在 NuGet 上，灵感来自 Stephen Toub's blog post series on the subject .我的原语都是同步和异步兼容的(和线程安全的)，这很有用，例如，资源的一个用户是同步的，而其他用户是异步的。