Asynchrone Datenströme

1. Einführung

Ihr kennt schon async und await. Die funktionieren großartig für "einzelne" asynchrone Aktionen, wie das Herunterladen einer Datei. Aber was, wenn die Daten als Strom ankommen oder eine Ressource eine asynchrone Bereinigung braucht?

Problem mit asynchronen "Collections": Stellt euch vor, ihr müsst Millionen von Einträgen aus einer Datenbank bekommen. Wenn eine Methode Task<List<T>> zurückgibt, wartet ihr, bis alle Daten im Speicher sind. Das ist ineffizient und verursacht Verzögerungen. Ein synchrones IEnumerable<T> ist auch ungeeignet, wenn jedes Element asynchron beschafft werden muss.

Problem mit asynchroner Ressourcenfreigabe: IDisposable und using sind gut für synchrone Bereinigung. Aber wie schaut's aus, wenn das Schließen einer Netzverbindung oder das Flushen von Puffern selbst asynchron ist? Man kann await nicht innerhalb eines synchronen Dispose() aufrufen, was zu Thread-Blocking oder falscher Bereinigung führt.

Um diese Probleme zu lösen, wurden IAsyncEnumerable<T> und IAsyncDisposable eingeführt.

2. Asynchrone Datenströme

IAsyncEnumerable<T> ist das asynchrone Gegenstück zu IEnumerable<T>. Es erlaubt, Elemente einer Sequenz asynchron eins nach dem anderen zu erzeugen, ohne darauf zu warten, dass alle Daten fertig sind.

Wann braucht man das?

• Asynchrones zeilenweises Lesen großer Dateien: z.B. Gigabyte-große Logs.
• Streaming von Daten aus dem Netzwerk oder der Datenbank: API-Ergebnisse, die in Teilen eintreffen.
• Implementierung serverseitiger Streaming-APIs: z.B. gRPC Streaming.
• In jedem Szenario, in dem Daten asynchron erzeugt oder empfangen werden und schrittweise verarbeitet werden sollen.

Wie funktioniert das?

1. IAsyncEnumerable<T>: Interface mit der Methode GetAsyncEnumerator(CancellationToken cancellationToken). Der Abbruch-Token ist wichtig!
2. IAsyncEnumerator<T>: Interface mit ValueTask<bool> MoveNextAsync() (Wechsel zum nächsten) und Current (aktuelles Element). Es erbt außerdem IAsyncDisposable.
3. await foreach: iteriert bequem über IAsyncEnumerable<T>. Der Compiler ruft automatisch MoveNextAsync() und Current auf. Wichtig: await foreach garantiert den Aufruf von DisposeAsync() für den Enumerator nach Abschluss der Iteration, auch bei Fehlern.

Erzeugen von IAsyncEnumerable<T> mit async yield return

Ihr könnt yield return innerhalb einer async-Methode verwenden, die IAsyncEnumerable<T> zurückgibt. Das erlaubt das Erstellen von asynchronen Generatoren. Eure Methode kann await verwenden, um die Erzeugung zu pausieren, auf eine asynchrone Operation zu warten und dann fortzufahren.

Beispiel: Einfacher asynchroner Generator

async IAsyncEnumerable<int> GenerateNumbersAsync()
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine($"Generiere: {i}");
        await Task.Delay(100); // Simulation asynchroner Arbeit
        yield return i; 
    }
}

// Verwendung:
async Task ConsumeAsyncNumbers()
{
    await foreach (var number in GenerateNumbersAsync())
    {
        Console.WriteLine($"Empfangen: {number}");
    }
}
// Aufrufen: await ConsumeAsyncNumbers();

Beispiel: Datei zeilenweise asynchron lesen

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

// Verwendung:
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($"Verarbeitete Zeile: {line}");
    }
}
// Aufrufen: await ProcessFileAsync();

Beispiel: Asynchroner Generator mit Abbruch (CancellationToken)

async IAsyncEnumerable<int> GetCancelableSequence(
    [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 0; i < 10; i++)
    {
        token.ThrowIfCancellationRequested(); // Prüfen auf Abbruch
        await Task.Delay(200, token); // Task.Delay unterstützt ebenfalls Abbruch über token
        yield return i;
    }
}

// Verwendung:
async Task ConsumeAndCancel()
{
    var cts = new CancellationTokenSource(500); // Abbruch nach 500ms
    try
    {
        await foreach (var num in GetCancelableSequence(cts.Token))
        {
            Console.WriteLine($"Empfangen: {num}");
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Erzeugung abgebrochen!");
    }
}
// Aufrufen: await ConsumeAndCancel();

Das Attribut [EnumeratorCancellation] erlaubt es, einen CancellationToken in den asynchronen Generator zu übergeben. So kann die Iteration abgebrochen werden, wenn der aufrufende Code über eine CancellationTokenSource Abbruch verlangt. Ohne dieses Attribut wird der Token nicht automatisch an GetAsyncEnumerator weitergereicht.

3. Asynchrone Ressourcenverwaltung

Problem des synchronen IDisposable

Die Methode Dispose() in IDisposable ist synchron (void Dispose()). Ihr könnt await darin nicht verwenden. Wenn das Schließen einer DB-Verbindung oder das Flushen von Puffern lange und asynchron ist, blockiert das synchrone Dispose() den Thread, was schlecht für asynchrone Anwendungen ist.

Lösung: IAsyncDisposable

IAsyncDisposable löst dieses Problem. Es enthält nur eine Methode: ValueTask DisposeAsync() — eine asynchrone Methode zur Bereinigung.

await using

Das ist das asynchrone Gegenstück zu using. Es ist für Objekte gedacht, die IAsyncDisposable implementieren.

• await using stellt sicher, dass DisposeAsync() am Ende des Codeblocks aufgerufen wird, in dem die Ressource deklariert wurde (oder beim Verlassen durch eine Ausnahme).
• Erlaubt korrekte asynchrone Freigabe von Ressourcen und vermeidet Blockierungen.

Beispiel: Einfaches IAsyncDisposable und await using

class MyAsyncResource : IAsyncDisposable
{
    public MyAsyncResource() => Console.WriteLine("Ressource geöffnet.");
    
    public async ValueTask DisposeAsync()
    {
        Console.WriteLine("Beginne asynchrone Bereinigung...");
        await Task.Delay(200); // Simulation asynchroner Bereinigung
        Console.WriteLine("Asynchrone Bereinigung abgeschlossen.");
    }
}

// Verwendung von await using
async Task UseAndDisposeResource()
{
    await using var resource = new MyAsyncResource(); 
    Console.WriteLine("Ressource wird benutzt...");
} // resource.DisposeAsync() wird hier automatisch aufgerufen

// Aufrufen: await UseAndDisposeResource();

Beispiel: Mehrere await using-Blöcke

async Task UseMultipleResources()
{
    await using var res1 = new MyAsyncResource();
    await using var res2 = new MyAsyncResource();
    Console.WriteLine("Benutze beide Ressourcen...");
} // Ressourcen werden in LIFO-Reihenfolge freigegeben: res2.DisposeAsync() wird zuerst aufgerufen, dann res1.DisposeAsync().
// Aufrufen: await UseMultipleResources();

Kompatibilität von IAsyncEnumerable<T> mit IAsyncDisposable

Wichtig: IAsyncEnumerator<T> (der von await foreach verwendet wird) erbt selbst IAsyncDisposable. Das bedeutet, wenn euer asynchroner Generator Ressourcen verwendet (wie den StreamReader im obigen Beispiel), die asynchron bereinigt werden können, kümmert sich await foreach darum. Es ruft DisposeAsync() des Enumerators auf, wenn die Iteration endet.