Semafor asinkron, kunci asinkron, dan koordinasi asinkron pembaca/penulis

Ketika kode asinkron membutuhkan pembatasan, eksklusi bersama, atau koordinasi pembaca/penulis, gunakan tipe .NET bawaan daripada membuat tipe Anda sendiri. Artikel ini menunjukkan cara menerapkan jenis tersebut, lalu menelusuri implementasi kustom untuk menjelaskan cara kerjanya secara internal.

Semaphore Asinkron — membatasi akses bersamaan

Semaphore membatasi berapa banyak penelepon yang dapat mengakses sumber daya secara bersamaan. SemaphoreSlim WaitAsync menyediakan sebuah metode yang memungkinkan Anda menunggu entri tanpa memblokir suatu utas.

public static class SemaphoreSlimDemo
{
    public static async Task RunAsync()
    {
        using var semaphore = new SemaphoreSlim(3);

        Task[] tasks = Enumerable.Range(1, 6).Select(id => Task.Run(async () =>
        {
            await semaphore.WaitAsync();
            try
            {
                Console.WriteLine($"Task {id}: entered (count = {semaphore.CurrentCount})");
                await Task.Delay(100);
            }
            finally
            {
                semaphore.Release();
                Console.WriteLine($"Task {id}: released");
            }
        })).ToArray();

        await Task.WhenAll(tasks);
    }
}

Public Module SemaphoreSlimDemo
    Public Async Function RunAsync() As Task
        Using semaphore As New SemaphoreSlim(3)
            Dim tasks As Task() = Enumerable.Range(1, 6).Select(
                Function(id) Task.Run(Async Function()
                    Await semaphore.WaitAsync()
                    Try
                        Console.WriteLine($"Task {id}: entered (count = {semaphore.CurrentCount})")
                        Await Task.Delay(100)
                    Finally
                        semaphore.Release()
                        Console.WriteLine($"Task {id}: released")
                    End Try
                End Function)).ToArray()

            Await Task.WhenAll(tasks)
        End Using
    End Function
End Module

Selalu pasangkan WaitAsync dengan Release dalam blok try/finally . Jika Anda lupa melepaskan, jumlah semaphore tidak pernah meningkat, dan penelepon lain menunggu tanpa batas waktu.

Cara kerja semaphore asinkron

Secara internal, asinkron semaphore mempertahankan hitungan dan antrean pelayan. Ketika hitungan di atas nol, WaitAsync deremen hitungan dan segera kembali. Saat jumlahnya nol, WaitAsync mengantrikan TaskCompletionSource dan kemudian mengembalikan tugasnya. Release menghapus antrean pelayan dan menyelesaikannya, atau menaikkan jumlah:

// Educational only — use SemaphoreSlim instead of this sample implementation.
public class AsyncSemaphore
{
    private readonly Queue<TaskCompletionSource> _waiters = new();
    private int _currentCount;

    public AsyncSemaphore(int initialCount)
    {
        ArgumentOutOfRangeException.ThrowIfNegative(initialCount, nameof(initialCount));
        _currentCount = initialCount;
    }

    public Task WaitAsync()
    {
        lock (_waiters)
        {
            if (_currentCount > 0)
            {
                _currentCount--;
                return Task.CompletedTask;
            }
            else
            {
                var waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
                _waiters.Enqueue(waiter);
                return waiter.Task;
            }
        }
    }

    public void Release()
    {
        TaskCompletionSource? toRelease = null;

        lock (_waiters)
        {
            if (_waiters.Count > 0)
                toRelease = _waiters.Dequeue();
            else
                _currentCount++;
        }

        toRelease?.TrySetResult();
    }
}

' Educational only — use SemaphoreSlim instead of this sample implementation.
Public Class AsyncSemaphore
    Private ReadOnly _waiters As New Queue(Of TaskCompletionSource)()
    Private _currentCount As Integer

    Public Sub New(initialCount As Integer)
        If initialCount < 0 Then Throw New ArgumentOutOfRangeException(NameOf(initialCount))
        _currentCount = initialCount
    End Sub

    Public Function WaitAsync() As Task
        SyncLock _waiters
            If _currentCount > 0 Then
                _currentCount -= 1
                Return Task.CompletedTask
            Else
                Dim waiter As New TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)
                _waiters.Enqueue(waiter)
                Return waiter.Task
            End If
        End SyncLock
    End Function

    Public Sub Release()
        Dim toRelease As TaskCompletionSource = Nothing

        SyncLock _waiters
            If _waiters.Count > 0 Then
                toRelease = _waiters.Dequeue()
            Else
                _currentCount += 1
            End If
        End SyncLock

        toRelease?.TrySetResult()
    End Sub
End Class

Metode Release ini melengkapi TaskCompletionSource di luar kunci, seperti halnya AsyncAutoResetEvent dalam membangun primitif koordinasi asinkron. Pendekatan ini mencegah kelanjutan sinkron berjalan saat kunci ditahan.

Kunci asinkron: penguncian-saling antar panggilan tunggu

Sebuah kunci dengan hitungan 1 menyediakan pengecualian saling. Pernyataan C# lock dan Lock (.NET 9+) tidak berfungsi di batas await karena bersifat thread-affine. Thread-affine berarti utas yang sama yang mengunci haruslah utas yang melepaskan kunci tersebut. awaitDi seluruh , utas yang melanjutkan kelanjutan mungkin bukan utas yang memperoleh kunci, yang melanggar persyaratan tersebut. Gunakan SemaphoreSlim dengan hitungan 1 sebagai gantinya:

public static class SemaphoreSlimAsLockDemo
{
    private static readonly SemaphoreSlim s_lock = new(1, 1);
    private static int s_sharedCounter;

    public static async Task RunAsync()
    {
        Task[] tasks = Enumerable.Range(1, 5).Select(_ => Task.Run(async () =>
        {
            await s_lock.WaitAsync();
            try
            {
                int before = s_sharedCounter;
                await Task.Delay(10);
                s_sharedCounter = before + 1;
            }
            finally
            {
                s_lock.Release();
            }
        })).ToArray();

        await Task.WhenAll(tasks);
        Console.WriteLine($"Counter = {s_sharedCounter} (expected 5)");
    }
}

Public Module SemaphoreSlimAsLockDemo
    Private ReadOnly s_lock As New SemaphoreSlim(1, 1)
    Private s_sharedCounter As Integer

    Public Async Function RunAsync() As Task
        Dim tasks As Task() = Enumerable.Range(1, 5).Select(
            Function(unused) Task.Run(Async Function()
                Await s_lock.WaitAsync()
                Try
                    Dim before As Integer = s_sharedCounter
                    Await Task.Delay(10)
                    s_sharedCounter = before + 1
                Finally
                    s_lock.Release()
                End Try
            End Function)).ToArray()

        Await Task.WhenAll(tasks)
        Console.WriteLine($"Counter = {s_sharedCounter} (expected 5)")
    End Function
End Module

Cara kerja kunci asinkron

Anda dapat membungkus pola semaphore dalam tipe yang mendukung using untuk pelepasan otomatis. Metode LockAsync mengembalikan Releaser sekali pakai; ketika Releaser dibuang, itu melepaskan semaphore.

// Educational only — use SemaphoreSlim(1, 1) with try/finally instead of this sample implementation.
public class AsyncLock : IDisposable
{
    private readonly SemaphoreSlim _semaphore = new(1, 1);
    private readonly Task<Releaser> _releaser;

    public AsyncLock()
    {
        _releaser = Task.FromResult(new Releaser(this));
    }

    public Task<Releaser> LockAsync()
    {
        Task wait = _semaphore.WaitAsync();
        return wait.IsCompleted
            ? _releaser
            : wait.ContinueWith(
                  (_, state) => new Releaser((AsyncLock)state!),
                  this,
                  CancellationToken.None,
                  TaskContinuationOptions.ExecuteSynchronously,
                  TaskScheduler.Default);
    }

    public struct Releaser : IDisposable
    {
        private readonly AsyncLock? _toRelease;

        internal Releaser(AsyncLock toRelease) => _toRelease = toRelease;

        public void Dispose() => _toRelease?._semaphore.Release();
    }

    public void Dispose() => _semaphore.Dispose();
}

' Educational only — use SemaphoreSlim(1, 1) with Try/Finally instead of this sample implementation.
Public Class AsyncLock
    Implements IDisposable

    Private ReadOnly _semaphore As New SemaphoreSlim(1, 1)
    Private ReadOnly _releaser As Task(Of Releaser)

    Public Sub New()
        _releaser = Task.FromResult(New Releaser(Me))
    End Sub

    Public Function LockAsync() As Task(Of Releaser)
        Dim wait As Task = _semaphore.WaitAsync()
        If wait.IsCompleted Then
            Return _releaser
        Else
            Return wait.ContinueWith(
                Function(unused, state) New Releaser(DirectCast(state, AsyncLock)),
                Me,
                CancellationToken.None,
                TaskContinuationOptions.ExecuteSynchronously,
                TaskScheduler.Default)
        End If
    End Function

    Public Structure Releaser
        Implements IDisposable

        Private ReadOnly _toRelease As AsyncLock

        Friend Sub New(toRelease As AsyncLock)
            _toRelease = toRelease
        End Sub

        Public Sub Dispose() Implements IDisposable.Dispose
            _toRelease?._semaphore.Release()
        End Sub
    End Structure

    Public Sub Dispose() Implements IDisposable.Dispose
        _semaphore.Dispose()
    End Sub
End Class

Penggunaan ringkas dan aman:

public static class AsyncLockDemo
{
    private static readonly AsyncLock s_lock = new();
    private static int s_sharedValue;

    public static async Task RunAsync()
    {
        Task[] tasks = Enumerable.Range(1, 5).Select(id => Task.Run(async () =>
        {
            using (await s_lock.LockAsync())
            {
                int before = s_sharedValue;
                await Task.Delay(10);
                s_sharedValue = before + 1;
                Console.WriteLine($"Task {id}: incremented to {s_sharedValue}");
            }
        })).ToArray();

        await Task.WhenAll(tasks);
        Console.WriteLine($"Final value = {s_sharedValue} (expected 5)");
    }
}

Public Module AsyncLockDemo
    Private ReadOnly s_lock As New AsyncLock()
    Private s_sharedValue As Integer

    Public Async Function RunAsync() As Task
        Dim tasks As Task() = Enumerable.Range(1, 5).Select(
            Function(id) Task.Run(Async Function()
                Using Await s_lock.LockAsync()
                    Dim before As Integer = s_sharedValue
                    Await Task.Delay(10)
                    s_sharedValue = before + 1
                    Console.WriteLine($"Task {id}: incremented to {s_sharedValue}")
                End Using
            End Function)).ToArray()

        Await Task.WhenAll(tasks)
        Console.WriteLine($"Final value = {s_sharedValue} (expected 5)")
    End Function
End Module

Koordinasi pembaca/penulis asinkron

Kunci pembaca/penulis memungkinkan beberapa pembaca bersamaan tetapi hanya satu penulis eksklusif. .NET menyediakan ConcurrentExclusiveSchedulerPair, yang menawarkan penjadwalan pembaca/penulis untuk tugas melalui dua instance TaskScheduler.

• ConcurrentScheduler — menjalankan tugas secara bersamaan (seperti pembaca), selama tidak ada tugas eksklusif yang aktif.
• ExclusiveScheduler — menjalankan tugas secara eksklusif (seperti penulis), tanpa tugas lain yang berjalan.

public static class ConcurrentExclusiveDemo
{
    public static async Task RunAsync()
    {
        var pair = new ConcurrentExclusiveSchedulerPair();
        var factory = new TaskFactory(pair.ExclusiveScheduler);

        int sharedValue = 0;

        Task writerTask = factory.StartNew(() =>
        {
            sharedValue = 42;
            Console.WriteLine($"Writer: set value to {sharedValue}");
        });

        var readerFactory = new TaskFactory(pair.ConcurrentScheduler);

        Task[] readerTasks = Enumerable.Range(1, 3).Select(id =>
            readerFactory.StartNew(() =>
            {
                Console.WriteLine($"Reader {id}: value = {sharedValue}");
            })).ToArray();

        await writerTask;
        await Task.WhenAll(readerTasks);
    }
}

Public Module ConcurrentExclusiveDemo
    Public Async Function RunAsync() As Task
        Dim pair As New ConcurrentExclusiveSchedulerPair()
        Dim exclusiveFactory As New TaskFactory(pair.ExclusiveScheduler)

        Dim sharedValue As Integer = 0

        Dim writerTask As Task = exclusiveFactory.StartNew(Sub()
            sharedValue = 42
            Console.WriteLine($"Writer: set value to {sharedValue}")
        End Sub)

        Dim readerFactory As New TaskFactory(pair.ConcurrentScheduler)

        Dim readerTasks As Task() = Enumerable.Range(1, 3).Select(
            Function(id) readerFactory.StartNew(Sub()
                Console.WriteLine($"Reader {id}: value = {sharedValue}")
            End Sub)).ToArray()

        Await writerTask
        Await Task.WhenAll(readerTasks)
    End Function
End Module

Kunci pembaca/penulis asinkron kustom

Implementasi berikut memberikan prioritas penulis atas pembaca. Ketika penulis menunggu, pembaca baru mengantre di belakangnya. Ketika penulis selesai dan tidak ada penulis lain yang menunggu, semua pembaca dalam antrean berjalan bersama.

public class AsyncReaderWriterLock
{
    private readonly Queue<TaskCompletionSource<Releaser>> _waitingWriters = new();
    private TaskCompletionSource<Releaser> _waitingReader =
        new(TaskCreationOptions.RunContinuationsAsynchronously);
    private int _readersWaiting;
    private int _status; // 0 = free, -1 = writer active, >0 = reader count

    private readonly Task<Releaser> _readerReleaser;
    private readonly Task<Releaser> _writerReleaser;

    public AsyncReaderWriterLock()
    {
        _readerReleaser = Task.FromResult(new Releaser(this, isWriter: false));
        _writerReleaser = Task.FromResult(new Releaser(this, isWriter: true));
    }

    public Task<Releaser> ReaderLockAsync()
    {
        lock (_waitingWriters)
        {
            if (_status >= 0 && _waitingWriters.Count == 0)
            {
                _status++;
                return _readerReleaser;
            }
            else
            {
                _readersWaiting++;
                return _waitingReader.Task;
            }
        }
    }

    public Task<Releaser> WriterLockAsync()
    {
        lock (_waitingWriters)
        {
            if (_status == 0)
            {
                _status = -1;
                return _writerReleaser;
            }
            else
            {
                var waiter = new TaskCompletionSource<Releaser>(TaskCreationOptions.RunContinuationsAsynchronously);
                _waitingWriters.Enqueue(waiter);
                return waiter.Task;
            }
        }
    }

    private void ReaderRelease()
    {
        TaskCompletionSource<Releaser>? toWake = null;

        lock (_waitingWriters)
        {
            _status--;
            if (_status == 0 && _waitingWriters.Count > 0)
            {
                _status = -1;
                toWake = _waitingWriters.Dequeue();
            }
        }

        toWake?.SetResult(new Releaser(this, isWriter: true));
    }

    private void WriterRelease()
    {
        TaskCompletionSource<Releaser>? toWake = null;
        bool toWakeIsWriter = false;

        lock (_waitingWriters)
        {
            if (_waitingWriters.Count > 0)
            {
                toWake = _waitingWriters.Dequeue();
                toWakeIsWriter = true;
            }
            else if (_readersWaiting > 0)
            {
                toWake = _waitingReader;
                _status = _readersWaiting;
                _readersWaiting = 0;
                _waitingReader = new TaskCompletionSource<Releaser>(TaskCreationOptions.RunContinuationsAsynchronously);
            }
            else
            {
                _status = 0;
            }
        }

        toWake?.SetResult(new Releaser(this, toWakeIsWriter));
    }

    public struct Releaser : IDisposable
    {
        private readonly AsyncReaderWriterLock? _lock;
        private readonly bool _isWriter;

        internal Releaser(AsyncReaderWriterLock lockObj, bool isWriter)
        {
            _lock = lockObj;
            _isWriter = isWriter;
        }

        public void Dispose()
        {
            if (_lock is not null)
            {
                if (_isWriter) _lock.WriterRelease();
                else _lock.ReaderRelease();
            }
        }
    }
}

Public Class AsyncReaderWriterLock
    Private ReadOnly _waitingWriters As New Queue(Of TaskCompletionSource(Of Releaser))()
    Private _waitingReader As New TaskCompletionSource(Of Releaser)(TaskCreationOptions.RunContinuationsAsynchronously)
    Private _readersWaiting As Integer
    Private _status As Integer ' 0 = free, -1 = writer active, >0 = reader count

    Private ReadOnly _readerReleaser As Task(Of Releaser)
    Private ReadOnly _writerReleaser As Task(Of Releaser)

    Public Sub New()
        _readerReleaser = Task.FromResult(New Releaser(Me, isWriter:=False))
        _writerReleaser = Task.FromResult(New Releaser(Me, isWriter:=True))
    End Sub

    Public Function ReaderLockAsync() As Task(Of Releaser)
        SyncLock _waitingWriters
            If _status >= 0 AndAlso _waitingWriters.Count = 0 Then
                _status += 1
                Return _readerReleaser
            Else
                _readersWaiting += 1
                Return _waitingReader.Task
            End If
        End SyncLock
    End Function

    Public Function WriterLockAsync() As Task(Of Releaser)
        SyncLock _waitingWriters
            If _status = 0 Then
                _status = -1
                Return _writerReleaser
            Else
                Dim waiter As New TaskCompletionSource(Of Releaser)(
                    System.Threading.Tasks.TaskCreationOptions.RunContinuationsAsynchronously)
                _waitingWriters.Enqueue(waiter)
                Return waiter.Task
            End If
        End SyncLock
    End Function

    Private Sub ReaderRelease()
        Dim toWake As TaskCompletionSource(Of Releaser) = Nothing

        SyncLock _waitingWriters
            _status -= 1
            If _status = 0 AndAlso _waitingWriters.Count > 0 Then
                _status = -1
                toWake = _waitingWriters.Dequeue()
            End If
        End SyncLock

        toWake?.SetResult(New Releaser(Me, isWriter:=True))
    End Sub

    Private Sub WriterRelease()
        Dim toWake As TaskCompletionSource(Of Releaser) = Nothing
        Dim toWakeIsWriter As Boolean = False

        SyncLock _waitingWriters
            If _waitingWriters.Count > 0 Then
                toWake = _waitingWriters.Dequeue()
                toWakeIsWriter = True
            ElseIf _readersWaiting > 0 Then
                toWake = _waitingReader
                _status = _readersWaiting
                _readersWaiting = 0
                _waitingReader = New TaskCompletionSource(Of Releaser)(TaskCreationOptions.RunContinuationsAsynchronously)
            Else
                _status = 0
            End If
        End SyncLock

        toWake?.SetResult(New Releaser(Me, toWakeIsWriter))
    End Sub

    Public Structure Releaser
        Implements IDisposable

        Private ReadOnly _lock As AsyncReaderWriterLock
        Private ReadOnly _isWriter As Boolean

        Friend Sub New(lockObj As AsyncReaderWriterLock, isWriter As Boolean)
            _lock = lockObj
            _isWriter = isWriter
        End Sub

        Public Sub Dispose() Implements IDisposable.Dispose
            If _lock IsNot Nothing Then
                If _isWriter Then
                    _lock.WriterRelease()
                Else
                    _lock.ReaderRelease()
                End If
            End If
        End Sub
    End Structure
End Class

Penggunaan mengikuti pola pelepasan sekali pakai yang sama dengan AsyncLock:

public static class AsyncReaderWriterLockDemo
{
    private static readonly AsyncReaderWriterLock s_rwLock = new();
    private static string s_data = "initial";

    public static async Task RunAsync()
    {
        Task writer = Task.Run(async () =>
        {
            using (await s_rwLock.WriterLockAsync())
            {
                Console.WriteLine("Writer: acquired exclusive lock");
                await Task.Delay(50);
                s_data = "updated";
                Console.WriteLine("Writer: data updated");
            }
        });

        Task[] readers = Enumerable.Range(1, 3).Select(id => Task.Run(async () =>
        {
            await Task.Delay(10);
            using (await s_rwLock.ReaderLockAsync())
            {
                Console.WriteLine($"Reader {id}: data = {s_data}");
            }
        })).ToArray();

        await writer;
        await Task.WhenAll(readers);
    }
}

Public Module AsyncReaderWriterLockDemo
    Private ReadOnly s_rwLock As New AsyncReaderWriterLock()
    Private s_data As String = "initial"

    Public Async Function RunAsync() As Task
        Dim writer As Task = Task.Run(Async Function()
            Using Await s_rwLock.WriterLockAsync()
                Console.WriteLine("Writer: acquired exclusive lock")
                Await Task.Delay(50)
                s_data = "updated"
                Console.WriteLine("Writer: data updated")
            End Using
        End Function)

        Dim readers As Task() = Enumerable.Range(1, 3).Select(
            Function(id) Task.Run(Async Function()
                Await Task.Delay(10)
                Using Await s_rwLock.ReaderLockAsync()
                    Console.WriteLine($"Reader {id}: data = {s_data}")
                End Using
            End Function)).ToArray()

        Await writer
        Await Task.WhenAll(readers)
    End Function
End Module

Saluran sebagai pola koordinasi alternatif

Channel<T> menyediakan antrean produsen-konsumen yang aman terhadap utas async yang mendukung baca dan tulis. Saluran dibatasi (CreateBounded) menghasilkan back-pressure secara alami, menggantikan beberapa skenario di mana Anda akan menggunakan semaphore untuk mengatur laju.

Untuk informasi selengkapnya, lihat System.Threading.Channels.

Baca juga

• Membangun primitif asinkron untuk koordinasi
• Menjaga metode asinkron tetap hidup
• Menyelesaikan tugas Anda
• Menggunakan Pola Asinkron Berbasis Tugas