Dòng dữ liệu bất đồng bộ

1. Giới thiệu

Bạn đã quen với async và await. Chúng hoạt động tốt cho các hành động bất đồng bộ "đơn lẻ", như tải một file. Nhưng còn khi dữ liệu đến theo dòng, hoặc tài nguyên cần giải phóng bất đồng bộ thì sao?

Vấn đề với "collection" bất đồng bộ: Hãy tưởng tượng cần lấy triệu bản ghi từ database. Nếu method trả về Task<List<T>>, bạn phải chờ cho đến khi tất cả dữ liệu load vào bộ nhớ. Điều này kém hiệu quả và gây delay. IEnumerable<T> đồng bộ cũng không phù hợp nếu mỗi phần tử phải được lấy bất đồng bộ.

Vấn đề với giải phóng tài nguyên bất đồng bộ: IDisposable và using xử lý tốt việc dọn dẹp đồng bộ. Nhưng nếu đóng kết nối mạng hay flush buffer vào file là hành động bất đồng bộ? Không thể gọi await trong Dispose() đồng bộ, dẫn tới block thread hoặc dọn dẹp sai.

Để giải quyết những vấn đề này, người ta giới thiệu IAsyncEnumerable<T> và IAsyncDisposable.

2. Dòng dữ liệu bất đồng bộ

IAsyncEnumerable<T> — là phiên bản bất đồng bộ của IEnumerable<T>. Nó cho phép sản xuất các phần tử của chuỗi một cách bất đồng bộ từng phần một, không cần chờ tất cả dữ liệu sẵn sàng.

Khi nào cần?

• Đọc file lớn theo dòng bất đồng bộ: ví dụ logs gigabyte.
• Streaming dữ liệu từ mạng hoặc database: kết quả API đến từng phần.
• Triển khai API streaming trên server: ví dụ gRPC Streaming.
• Bất kỳ kịch bản nào mà dữ liệu được sinh hoặc đến bất đồng bộ và cần xử lý từng phần.

Nó hoạt động thế nào?

1. IAsyncEnumerable<T>: interface có method GetAsyncEnumerator(CancellationToken cancellationToken). Cancellation token rất quan trọng!
2. IAsyncEnumerator<T>: interface có ValueTask<bool> MoveNextAsync() (chuyển sang phần tử tiếp theo) và Current (phần tử hiện tại). Nó cũng kế thừa IAsyncDisposable.
3. await foreach: tiện để lặp qua IAsyncEnumerable<T>. Compiler tự gọi MoveNextAsync() và truy cập Current. Quan trọng là await foreach đảm bảo gọi DisposeAsync() cho iterator khi kết thúc lặp, kể cả khi có lỗi.

Tạo IAsyncEnumerable<T> bằng async yield return

Bạn có thể dùng yield return trong method async trả về IAsyncEnumerable<T>. Điều này cho phép tạo generator bất đồng bộ. Method có thể dùng await để tạm dừng sinh, chờ thao tác bất đồng bộ rồi tiếp tục.

Ví dụ: Generator bất đồng bộ đơn giản

async IAsyncEnumerable<int> GenerateNumbersAsync()
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine($"Generiruyu: {i}");
        await Task.Delay(100); // Imitatsiya asinkhronnoy raboty
        yield return i; 
    }
}

// Ispol'zovanie:
async Task ConsumeAsyncNumbers()
{
    await foreach (var number in GenerateNumbersAsync())
    {
        Console.WriteLine($"Polucheno: {number}");
    }
}
// Vyzovite: await ConsumeAsyncNumbers();

Ví dụ: Đọc file theo dòng bất đồng bộ

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

// Ispol'zovanie:
async Task ProcessFileAsync()
{
    await File.WriteAllLinesAsync("data.txt", new[] { "Stroka 1", "Stroka 2", "Stroka 3" });
    await foreach (var line in ReadFileLinesAsync("data.txt")) 
    {
        Console.WriteLine($"Obrabotana stroka: {line}");
    }
}
// Vyzovite: await ProcessFileAsync();

Ví dụ: Generator bất đồng bộ có huỷ (CancellationToken)

async IAsyncEnumerable<int> GetCancelableSequence(
    [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 0; i < 10; i++)
    {
        token.ThrowIfCancellationRequested(); // Proveryaem otmenu
        await Task.Delay(200, token); // Task.Delay takzhe podderzhivaet otmenu cherez token
        yield return i;
    }
}

// Ispol'zovanie:
async Task ConsumeAndCancel()
{
    var cts = new CancellationTokenSource(500); // Otmena cherez 500ms
    try
    {
        await foreach (var num in GetCancelableSequence(cts.Token))
        {
            Console.WriteLine($"Polucheno: {num}");
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Generaciya otmenena!");
    }
}
// Vyzovite: await ConsumeAndCancel();

Attribute [EnumeratorCancellation] cho phép truyền CancellationToken vào generator bất đồng bộ. Điều này cho phép huỷ vòng lặp nếu code gọi yêu cầu huỷ qua CancellationTokenSource. Nếu không có attribute này thì token sẽ không tự động được truyền vào GetAsyncEnumerator.

3. Quản lý tài nguyên bất đồng bộ

Vấn đề của IDisposable đồng bộ

Method Dispose() trong IDisposable là đồng bộ (void Dispose()). Bạn không thể dùng await trong đó. Nếu đóng kết nối DB hoặc flush buffer vào file là thao tác dài và bất đồng bộ, Dispose() đồng bộ sẽ block thread, điều này không tốt cho app bất đồng bộ.

Giải pháp: IAsyncDisposable

IAsyncDisposable giải quyết vấn đề này. Nó chứa một method duy nhất: ValueTask DisposeAsync() — method bất đồng bộ để dọn dẹp.

await using

Đây là phiên bản bất đồng bộ của using. Dùng cho các object implement IAsyncDisposable.

• await using đảm bảo gọi DisposeAsync() khi kết thúc khối code nơi resource được khai báo (hoặc khi thoát do exception).
• Cho phép giải phóng tài nguyên một cách bất đồng bộ, tránh block.

Ví dụ: IAsyncDisposable cơ bản và await using

class MyAsyncResource : IAsyncDisposable
{
    public MyAsyncResource() => Console.WriteLine("Resurs otkryt.");
    
    public async ValueTask DisposeAsync()
    {
        Console.WriteLine("Nachinayu asinhronnuyu ochistku...");
        await Task.Delay(200); // Imitatsiya asinhronnoy ochistki
        Console.WriteLine("Asinhronnaya ochistka zavershena.");
    }
}

// Ispol'zovanie await using
async Task UseAndDisposeResource()
{
    await using var resource = new MyAsyncResource(); 
    Console.WriteLine("Resurs ispol'zuetsya...");
} // Zdes' avtomaticheski vyzyvaetsya resource.DisposeAsync()

// Vyzovite: await UseAndDisposeResource();

Ví dụ: Nhiều khối await using

async Task UseMultipleResources()
{
    await using var res1 = new MyAsyncResource();
    await using var res2 = new MyAsyncResource();
    Console.WriteLine("Ispol'zuju oba resursa...");
} // resursy osvobozhdayutsya v poriadke LIFO (Last In, First Out): res2.DisposeAsync() vyzyvaetsya pervym, zatem res1.DisposeAsync().
// Vyzovite: await UseMultipleResources();

Tương thích giữa IAsyncEnumerable<T> và IAsyncDisposable

Quan trọng: IAsyncEnumerator<T> (được dùng bởi await foreach) tự nó kế thừa IAsyncDisposable. Điều này nghĩa là nếu generator bất đồng bộ của bạn sử dụng tài nguyên (như StreamReader trong ví dụ phía trên) có thể được dọn dẹp bất đồng bộ, thì await foreach sẽ lo chuyện đó. Nó sẽ gọi DisposeAsync() trên iterator khi vòng lặp kết thúc.