Asynchroniczne strumienie danych

1. Wprowadzenie

Znacie już async i await. Działają świetnie dla "pojedynczych" asynchronicznych operacji, jak pobranie jednego pliku. Ale co jeśli dane przychodzą strumieniowo, albo zasób wymaga asynchronicznego czyszczenia?

Problem z asynchronicznymi "kolekcjami": Wyobraźcie sobie, że trzeba pobrać miliony rekordów z bazy. Jeśli metoda zwraca Task<List<T>>, czekacie aż wszystkie dane załadują się do pamięci. To nieefektywne i powoduje opóźnienia. Synchronous IEnumerable<T> też nie pasuje, jeśli każdy element trzeba pobierać asynchronicznie.

Problem z asynchronicznym zwalnianiem zasobów: IDisposable i using radzą sobie z synchronicznym czyszczeniem. Ale co, gdy zamknięcie połączenia sieciowego lub zapis buforów do pliku jest samo w sobie asynchroniczne? Nie można wywołać await wewnątrz synchronicznego Dispose(), co prowadzi do blokowania wątku lub niepoprawnego czyszczenia.

Aby rozwiązać te problemy wprowadzono IAsyncEnumerable<T> i IAsyncDisposable.

2. Asynchroniczne strumienie danych

IAsyncEnumerable<T> — to asynchroniczny odpowiednik IEnumerable<T>. Pozwala asynchronicznie produkować elementy sekwencji jeden po drugim, bez czekania aż wszystkie dane będą gotowe.

Kiedy to potrzebne?

• Asynchroniczne czytanie dużych plików linia po linii: np. gigabajtowe logi.
• Streaming danych z sieci lub bazy danych: wyniki zapytań do API, które przychodzą partiami.
• Implementacja serwerowych API strumieniowych: np. gRPC Streaming.
• W każdym scenariuszu, gdzie dane generowane są lub przychodzą asynchronicznie i powinny być przetwarzane stopniowo.

Jak to działa?

1. IAsyncEnumerable<T>: interfejs z metodą GetAsyncEnumerator(CancellationToken cancellationToken). Token anulowania jest ważny!
2. IAsyncEnumerator<T>: interfejs z ValueTask<bool> MoveNextAsync() (przejście do następnego) i Current (bieżący element). On również dziedziczy IAsyncDisposable.
3. await foreach: wygodnie iteruje po IAsyncEnumerable<T>. Kompilator sam wywołuje MoveNextAsync() i Current. Najważniejsze, await foreach gwarantuje wywołanie DisposeAsync() dla enumeratora po zakończeniu iteracji, nawet przy błędach.

Tworzenie IAsyncEnumerable<T> z async yield return

Możesz użyć yield return wewnątrz async-metody, która zwraca IAsyncEnumerable<T>. To pozwala tworzyć asynchroniczne generatory. Twoja metoda może używać await, żeby wstrzymać generowanie, poczekać na asynchroniczną operację, a potem wznowić pracę.

Przykład: Prosty asynchroniczny generator

async IAsyncEnumerable<int> GenerateNumbersAsync()
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine($"Generuję: {i}");
        await Task.Delay(100); // Symulacja asynchronicznej pracy
        yield return i; 
    }
}

// Użycie:
async Task ConsumeAsyncNumbers()
{
    await foreach (var number in GenerateNumbersAsync())
    {
        Console.WriteLine($"Otrzymano: {number}");
    }
}
// Wywołaj: await ConsumeAsyncNumbers();

Przykład: Asynchroniczne czytanie pliku linia po linii

async IAsyncEnumerable<string> ReadFileLinesAsync(string filePath)
{
    using var reader = new StreamReader(filePath); // 'using' tutaj (StreamReader implementuje IAsyncDisposable)
    string? line;
    while ((line = await reader.ReadLineAsync()) != null) 
    {
        yield return line;
    }
}

// Użycie:
async Task ProcessFileAsync()
{
    await File.WriteAllLinesAsync("data.txt", new[] { "String 1", "String 2", "String 3" });
    await foreach (var line in ReadFileLinesAsync("data.txt")) 
    {
        Console.WriteLine($"Przetworzono linię: {line}");
    }
}
// Wywołaj: await ProcessFileAsync();

Przykład: Asynchroniczny generator z anulowaniem (CancellationToken)

async IAsyncEnumerable<int> GetCancelableSequence(
    [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 0; i < 10; i++)
    {
        token.ThrowIfCancellationRequested(); // Sprawdzamy anulowanie
        await Task.Delay(200, token); // Task.Delay też wspiera anulowanie przez token
        yield return i;
    }
}

// Użycie:
async Task ConsumeAndCancel()
{
    var cts = new CancellationTokenSource(500); // Anulowanie po 500ms
    try
    {
        await foreach (var num in GetCancelableSequence(cts.Token))
        {
            Console.WriteLine($"Otrzymano: {num}");
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Generowanie anulowane!");
    }
}
// Wywołaj: await ConsumeAndCancel();

Atrybut [EnumeratorCancellation] pozwala przekazać CancellationToken do asynchronicznego generatora. Daje to możliwość anulowania iteracji, jeśli kod wywołujący poprosi o anulowanie przez CancellationTokenSource. Bez tego atrybutu token nie będzie automatycznie przekazany do GetAsyncEnumerator.

3. Asynchroniczne zarządzanie zasobami

Problem synchronicznego IDisposable

Metoda Dispose() w IDisposable jest synchroniczna (void Dispose()). Nie możesz użyć await w jej wnętrzu. Jeśli zamknięcie połączenia z DB albo zrzut buforów do pliku to długie asynchroniczne operacje, synchroniczne Dispose() zablokuje wątek, co jest złe dla aplikacji asynchronicznych.

Rozwiązanie: IAsyncDisposable

IAsyncDisposable rozwiązuje ten problem. Zawiera pojedynczą metodę: ValueTask DisposeAsync() — asynchroniczną metodę do czyszczenia.

await using

To asynchroniczny odpowiednik using. Jest przeznaczony dla obiektów implementujących IAsyncDisposable.

• await using gwarantuje, że DisposeAsync() zostanie wywołany po zakończeniu bloku kodu, w którym zadeklarowano zasób (albo przy wyjściu z niego wskutek wyjątku).
• Pozwala poprawnie zwalniać zasoby asynchronicznie, unikając blokad.

Przykład: Podstawowe IAsyncDisposable i await using

class MyAsyncResource : IAsyncDisposable
{
    public MyAsyncResource() => Console.WriteLine("Zasób otwarty.");
    
    public async ValueTask DisposeAsync()
    {
        Console.WriteLine("Zaczynam asynchroniczne czyszczenie...");
        await Task.Delay(200); // Symulacja asynchronicznego czyszczenia
        Console.WriteLine("Asynchroniczne czyszczenie zakończone.");
    }
}

// Użycie await using
async Task UseAndDisposeResource()
{
    await using var resource = new MyAsyncResource(); 
    Console.WriteLine("Zasób jest używany...");
} // Tutaj automatycznie wywoływane jest resource.DisposeAsync()

// Wywołaj: await UseAndDisposeResource();

Przykład: Kilka bloków await using

async Task UseMultipleResources()
{
    await using var res1 = new MyAsyncResource();
    await using var res2 = new MyAsyncResource();
    Console.WriteLine("Używam obu zasobów...");
} // zasoby są zwalniane w porządku LIFO (Last In, First Out): res2.DisposeAsync() wywoływane jest pierwsze, potem res1.DisposeAsync().
// Wywołaj: await UseMultipleResources();

Kompatybilność IAsyncEnumerable<T> z IAsyncDisposable

Ważne: IAsyncEnumerator<T> (który używa await foreach) sam dziedziczy IAsyncDisposable. To znaczy, że jeśli twój asynchroniczny generator używa zasobów (jak StreamReader w przykładzie wyżej), które można czyścić asynchronicznie, await foreach zadba o to. Wywoła DisposeAsync() na enumeratorze, gdy iteracja się zakończy.