Oto bardzo podstawowa implementacja, BlockingCollection
która obsługuje oczekiwanie, z wieloma brakującymi funkcjami. Korzysta z AsyncEnumerable
biblioteki, która umożliwia asynchroniczne wyliczanie dla wersji C # starszych niż 8,0.
public class AsyncBlockingCollection<T>
{
private Queue<T> _queue = new Queue<T>();
private SemaphoreSlim _semaphore = new SemaphoreSlim(0);
private int _consumersCount = 0;
private bool _isAddingCompleted;
public void Add(T item)
{
lock (_queue)
{
if (_isAddingCompleted) throw new InvalidOperationException();
_queue.Enqueue(item);
}
_semaphore.Release();
}
public void CompleteAdding()
{
lock (_queue)
{
if (_isAddingCompleted) return;
_isAddingCompleted = true;
if (_consumersCount > 0) _semaphore.Release(_consumersCount);
}
}
public IAsyncEnumerable<T> GetConsumingEnumerable()
{
lock (_queue) _consumersCount++;
return new AsyncEnumerable<T>(async yield =>
{
while (true)
{
lock (_queue)
{
if (_queue.Count == 0 && _isAddingCompleted) break;
}
await _semaphore.WaitAsync();
bool hasItem;
T item = default;
lock (_queue)
{
hasItem = _queue.Count > 0;
if (hasItem) item = _queue.Dequeue();
}
if (hasItem) await yield.ReturnAsync(item);
}
});
}
}
Przykład użycia:
var abc = new AsyncBlockingCollection<int>();
var producer = Task.Run(async () =>
{
for (int i = 1; i <= 10; i++)
{
await Task.Delay(100);
abc.Add(i);
}
abc.CompleteAdding();
});
var consumer = Task.Run(async () =>
{
await abc.GetConsumingEnumerable().ForEachAsync(async item =>
{
await Task.Delay(200);
await Console.Out.WriteAsync(item + " ");
});
});
await Task.WhenAll(producer, consumer);
Wynik:
1 2 3 4 5 6 7 8 9 10
Aktualizacja: wraz z wydaniem języka C # 8 asynchroniczne wyliczanie stało się funkcją języka wbudowanego. Wymagane klasy ( IAsyncEnumerable
, IAsyncEnumerator
) są osadzone w .NET Core 3.0 i są oferowane jako pakiet dla .NET Framework 4.6.1+ ( Microsoft.Bcl.AsyncInterfaces ).
Oto alternatywna GetConsumingEnumerable
implementacja z nową składnią C # 8:
public async IAsyncEnumerable<T> GetConsumingEnumerable()
{
lock (_queue) _consumersCount++;
while (true)
{
lock (_queue)
{
if (_queue.Count == 0 && _isAddingCompleted) break;
}
await _semaphore.WaitAsync();
bool hasItem;
T item = default;
lock (_queue)
{
hasItem = _queue.Count > 0;
if (hasItem) item = _queue.Dequeue();
}
if (hasItem) yield return item;
}
}
Zwróć uwagę na współistnienie await
i yield
w tej samej metodzie.
Przykład użycia (C # 8):
var consumer = Task.Run(async () =>
{
await foreach (var item in abc.GetConsumingEnumerable())
{
await Task.Delay(200);
await Console.Out.WriteAsync(item + " ");
}
});
Zwróć uwagę na await
przed foreach
.
await Task.Run(() => blockingCollection.Take())
zadania, zostanie ono wykonane w innym wątku, a Twój wątek interfejsu użytkownika nie zostanie zablokowany, czy nie o to chodzi?