Что такое async streams (IAsyncEnumerable<T>) и когда они лучше Task<List<T>>?

Простой пример: чтение строк файла

// Производитель
public async IAsyncEnumerable<string> ReadLinesAsync(
    string path,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    using var reader = new StreamReader(path);
    while (!reader.EndOfStream)
    {
        ct.ThrowIfCancellationRequested();
        var line = await reader.ReadLineAsync(ct);
        if (line is not null)
            yield return line;
    }
}

// Потребитель
await foreach (var line in ReadLinesAsync("data.csv", cts.Token))
{
    Process(line);
}

Async Streams — IAsyncEnumerable<T>

Async streams — это механизм для асинхронного итеративного получения данных. Метод объявляется как async IAsyncEnumerable<T>, использует yield return и потребляется через await foreach. Каждый элемент доступен сразу после появления, без ожидания полной коллекции.

Стриминг из базы данных (EF Core)

// EF Core поддерживает AsAsyncEnumerable() для стриминга без загрузки всей таблицы
public async IAsyncEnumerable<Order> GetLargeOrdersAsync(
    decimal minAmount,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    await foreach (var order in _db.Orders
        .Where(o => o.Amount > minAmount)
        .AsAsyncEnumerable()
        .WithCancellation(ct))
    {
        yield return order;
    }
}

Когда IAsyncEnumerable<T> лучше Task<List<T>>

• Большой набор данных — не нужно держать всё в памяти одновременно.
• Стриминг к клиенту — первые результаты отдаются до завершения всей операции (Minimal API с IAsyncEnumerable поддерживается напрямую).
• Данные поступают постепенно — очереди сообщений, Kafka, WebSocket.
• Ранняя отмена — потребитель может остановиться после первых N элементов.

Minimal API — нативная поддержка

app.MapGet("/events", (IEventService svc, CancellationToken ct) =>
    svc.StreamEventsAsync(ct)); // возвращает IAsyncEnumerable, ASP.NET Core стримит JSON-массив

Channel как буфер между производителем и потребителем

public async IAsyncEnumerable<int> ProduceAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    var channel = Channel.CreateBounded<int>(100);

    _ = Task.Run(async () =>
    {
        for (int i = 0; i < 1000; i++)
            await channel.Writer.WriteAsync(i, ct);
        channel.Writer.Complete();
    }, ct);

    await foreach (var item in channel.Reader.ReadAllAsync(ct))
        yield return item;
}

Подводные камни

• Забыть атрибут [EnumeratorCancellation] — токен, переданный через .WithCancellation(ct), не попадёт в метод.
• Использовать IAsyncEnumerable для маленьких коллекций — избыточный overhead state-machine без реальной пользы.
• Не диспозить перечислитель явно при ручном вызове GetAsyncEnumerator() — использовать await foreach или ConfigureAwait на уровне foreach.
• Вызывать ToListAsync() на IAsyncEnumerable — теряет весь смысл стриминга.
• Смешивать yield return с try/finally без понимания: блок finally в async-итераторе выполняется при диспозе перечислителя, а не сразу.
• Параллельный перебор одного IAsyncEnumerable из нескольких потоков — не потокобезопасно, нужен Channel или SemaphoreSlim.

• Отвечать определением без production-сценария.
• Не называть runtime boundary, security boundary или failure mode.
• Игнорировать версию API, observability и тестовую проверку.

• Объясняет механизм своими словами и без выдуманных API.
• Называет реальные риски, диагностику и критерий корректности.
• Связывает ответ с текущей документацией и миграционными ограничениями.