Недавно мне попалась отличная статья про IAsyncEnumerable и стриминг данных. В ней у автора упал прод, который пытался выдать 500 000 записей разом и упал на вызове ToListAsync() с OOM при 8 ГБ RAM. Далее в статье описывается, как все это стримить с помощью IAsyncEnumerable с примерами кода. В целом после прочтения статьи может сложиться впечатление, что все свои ToListAsync() срочно нужно убрать и заменить на стриминг.

Но со времен появления стримингового апи мне всегда было скорее интересно:

1. Как отдавать стримы на фронтенд?

2. В каких случаях это стоит делать?

3. Как их принимать на фронтенде?

Но руки до этих вопросов упорно не доходили, а статья послужила эдаким триггером к изучению, результатами которого я и хочу поделиться.

Как стриминг работает: концепция

Для начала — в чем вообще идея стриминга данных и почему он действительно может помогать экономить память в сравнении с ToListAsync()?

Возьмем, к примеру, вот такой код:

app.MapGet("/articles", async (KbContext db, CancellationToken ct) =>
           Results.Ok(await db.Articles.AsNoTracking().ToListAsync(ct)));

Когда вы делаете ToListAsync(), то в памяти это иллюстративно выглядит вот так:

1. Сначала копируем все 500 000 записей из БД в память процесса. 

2. А затем отдаем клиенту в сокет. 

По итогу мы уже, вроде, отдали клиенту все, что надо, и то, что осталось в памяти процесса — нам больше не нужно. Но оно раздуло нам память, и будет висеть там до следующей сборки мусора. Это плохо.

Стриминг в этом плане выглядит куда более интересно. Вы перекачиваете записи по одной и отдаете в сокет. Например, достали одну запись и положили к себе в память, а затем — отдали клиенту. Взяли следующую — и так далее.

Но как добиться того, чтобы отдача шла по одной записи, а не огромной пачкой по 500 000 объектов за раз?

Способ 1: отдавать IAsyncEnumerable напрямую

Самый ленивый стриминг. ASP.NET Core умеет сериализовать IAsyncEnumerable<T> как стриминговый JSON-массив — System.Text.Json и пишет элементы по мере перечисления:

Бэкенд

app.MapGet("/articles", (IDbContextFactory<KbContext> factory, CancellationToken ct)
           => StreamArticles(factory, ct));

static async IAsyncEnumerable<Article> StreamArticles(
    IDbContextFactory<KbContext> factory,
    [EnumeratorCancellation] CancellationToken ct)
{
    await using var db = await factory.CreateDbContextAsync(ct);

    var articles = db.Articles.AsNoTracking()
                       .AsAsyncEnumerable()
                       .WithCancellation(ct);

    await foreach (var a in articles)
        yield return a;
}

Этот подход частично сработает. Такой код будет выдавать JSON-массив в таком формате: [{...},{...}]

Фронтенд

Тут сработает самый обычный fetch:

const res = await fetch("/articles");
const articles = await res.json(); // придётся дождаться всего массива целиком

articles.forEach(render);
res.json();

Для того, чтобы прочитать данные, фронтенду нужно знать, где заканчивается массив. А значит, фронт будет выкачивать массив до конца и не сможет рендерить промежуточные результаты. В целом тоже рабочий способ: на сервере память при этом экономится (мы не держим весь List в куче, объекты уходят по одному) — такой вариант подходит, если нужно отдать большой массив данных, не раздувая память процесса, но рендерить по мере прихода данных на фронтенд не требуется.

Способ 2: NDJSON — разделить массив на строки

NDJSON (Newline-Delimited JSON) — по сути, это массив, но вместо того, чтобы посылать его весь разом [{...},{...}], мы каждую строку посылаем отдельно через перенос строки вот так: {...}\n{...}\n{...}. Так отдают логи, дампы, bulk-эндпоинты Elasticsearch и часть стримов LLM.

Бэкенд

Чтобы отдать стрим в таком формате, вам нужно написать на бэке что-то типа такого:

app.MapGet("/articles/pipeline", async (HttpContext http,
    IDbContextFactory<KbContext> factory, CancellationToken ct) =>
{
    http.Response.ContentType = "application/x-ndjson";

    await using var db = await factory.CreateDbContextAsync(ct);

    var articles = db.Articles.AsNoTracking() // без этого change tracker удержит всё в памяти
                     .AsAsyncEnumerable()
                     .WithCancellation(ct);

    // Шаги 1–3 спрятаны вот в этом await foreach.
    await foreach (var a in articles)
    {
        // [4] Пишем каждую строку в ответ
        // Сериализуем объект →
        await JsonSerializer.SerializeAsync(http.Response.Body, a, cancellationToken: ct);
        // Дописываем \n границу объекта для NDJSON
        await http.Response.Body.WriteAsync("\n"u8.ToArray(), ct);

        // [5] FlushAsync говорит Kestrel: "отдавай то, что накопилось, прямо сейчас".
        //     По HTTP/1.1 это отправится HTTP-чанком (chunked transfer encoding),
        //     по HTTP/2 — DATA-фреймом.
        await http.Response.Body.FlushAsync(ct);
    }
});

Рассмотрим по шагам, что физически происходит, когда мы стримим список таким способом:

Дальше детали будут про MySQL (она у меня в рабочем проекте), некоторые из них — особенности именно её протокола. Общая механика: БД отдаёт строки по мере чтения, а драйвер собирает их по одной — справедлива для большинства провайдеров. Но стримит ли конкретный драйвер по строке или втихую буферит весь результат — это уже зависит от него и его настроек, и проверять этот момент стоит отдельно.

1. MySQL шлёт строки результата по одной, отдельными пакетами: в text-протоколе каждая строка — это отдельный протокольный пакет. Обычно сервер льёт их по мере чтения; но если плану нужна сортировка без индекса или temp table, он сперва материализует весь результат, а уже потом отдаёт. «Пакет» тут — единица протокола MySQL, а не TCP-сегмент: TCP под ним волен и склеить несколько строк в один сегмент, и разрезать большую строку на несколько.

2. MySqlConnector читает эти пакеты из сокета по мере того, как ты просишь следующую строку, и собирает из байтов готовый объект. То есть, стриминг из базы есть всегда — ToListAsync просто вычитывает все строки разом и складывает в List. Скорость отдачи регулируется с помощью механизма backpressure (обратное давление), и за него отвечает уже протокол TCP: MySQL сервер пушит строки так быстро, как может, и если клиент (ваш бэк) не читает — забивается буфер сокета, и MySQL сервер встаёт на записи, ожидая пока чтение не продолжится.

3. EF Core через AsAsyncEnumerable() отдаёт тебе строки по одной (await foreach).

4. Ты пишешь каждую строку в ответ и при желании флашишь. На моменте флашинга ты, по сути, говоришь, что объект тебе больше не нужен, и его можно будет собрать при следующей сборке. В отличие от листа, который будет копиться и переживать все сборки пока не будет отдан полностью. То есть, важно, что мы все равно тут аллоцируем. Просто это будет маленький кусочек, который легко чистится. С EntityFramework без аллокаций тут не получится.

5. Kestrel отдаёт тело по кускам по мере записи.

6. Браузер получает байты по мере прихода. fetch отдаёт их через response.body — это поток (ReadableStream). В этом потоке нам самим придется ловить переносы строк \n 

Фронтенд

Ну и код, который будет все это принимать. На нем стоит остановиться отдельно:

const res = await fetch("/articles/pipeline");

// [6] Браузер отдаёт тело ответа по мере прихода байтов.
//     res.body — это ReadableStream, читаем его кусками через reader.

const reader = res.body!.getReader();

const decoder = new TextDecoder();

let buffer = "";

while (true) {
  const { done, value } = await reader.read(); // value — очередной сетевой чанк (НЕ строка!)
  if (done) break;

  buffer += decoder.decode(value, { stream: true });

  // Границы сетевых чанков не совпадают с границами объектов,
  // поэтому режем буфер по \n и рендерим только целые строки.
  let nl: number;
  while ((nl = buffer.indexOf("\n")) >= 0) {
    const line = buffer.slice(0, nl);
    buffer = buffer.slice(nl + 1);

    if (line.trim()) render(JSON.parse(line)); // дорезали один объект и можно рендерить
  }
}

Код выше придется написать руками, так как в JS встроенного парсера для NDJSON нет. Ну или можно будет найти готовую библиотеку.

Плюсы у такого подхода: работает где угодно — обычный POST, любые заголовки, любой HTTP-клиент (мобилка, бэк-ту-бэк, скрипт). Формат человекочитаемый, дебажится curl-ом.

Минусы: нет встроенного парсера и авто-реконнекта;

Способ 3: SSE — самый нативный вариант для браузера

Server-Sent Events — это одно долгоживущее HTTP-соединение, по которому сервер отсылает события. Формат кадра стандартизирован (data: ...\n\n, плюс необязательные event:, id:, retry:), и — главное — в браузере уже есть парсер этого формата.

Бэкенд

В .NET 10 это, по сути, одна строчка — TypedResults.ServerSentEvents. Хелпер сам сериализует объект в data: и оборачивает в SSE-формат:

app.MapGet("/articles", (IDbContextFactory<KbContext> factory, CancellationToken ct) => TypedResults.ServerSentEvents(Stream(factory, ct)));

static async IAsyncEnumerable<SseItem<Article?>> Stream(
    IDbContextFactory<KbContext> factory,
    [EnumeratorCancellation] CancellationToken ct)

{

    await using var db = await factory.CreateDbContextAsync(ct);

    var articles = db.Articles
        .AsNoTracking()
        .AsAsyncEnumerable()
        .WithCancellation(ct);

    await foreach (var a in articles)

    {
        yield return new SseItem<Article?>(a, "article");
    }

    yield return new SseItem<Article?>(null, "done"); // говорим что все отдали
}

Фронтенд

Тут все довольно нативно, поэтому код получится совсем простым и приятным, так как это поддерживаемый браузерами стандарт:

const es = new EventSource("/articles");

// Добавляем слушателя на событие
es.addEventListener("article", (e) => render(JSON.parse(e.data)));

// И на окончание потока событий, иначе EventSource переподключится.
es.addEventListener("done", () => es.close());

Плюсы: встроенный парсер в браузере, именованные события (article / done), авто-реконнект.

Минусы и грабли:

• EventSource умеет только GET без кастомных заголовков. Нужен POST или токен в заголовке — берите @microsoft/fetch-event-source.

• Авто-реконнект — палка о двух концах: когда стрим закончился, надо явно сказать клиенту «всё» (событие done + es.close()), иначе он переподключится и начнёт качать заново.

• Не забудьте CancellationToken. Соединение живёт долго; если клиент отвалился, а токен не проброшен, сервер продолжит тянуть строки из БД вникуда. В minimal API токен прилетает в хендлер автоматически (это RequestAborted).

• SSE не поддерживается в остальных платформах: на мобилках (iOS/Android), в бэк-ту-бэк и в скриптах формат text/event-stream придётся парсить руками или брать библиотеку. В общем, если потребитель — не браузер, часто проще использовать NDJSON.

Замер: буфер против стримов

Теперь посмотрим, что у разных вариантов решений по памяти. 

Я попробовал прогнать на k6: 8 виртуальных пользователей, 12 секунд непрерывной нагрузки, ответ ~47 МБ у всех трёх вариантов одинаковый.

Пик памяти сервера снимал из встроенного семплера кучи.

Вот тут можно посмотреть как это сделано и запустить у себя (нужно ссылку на репу)

• Память у стримов в разы меньше. Оно и логично. Мы просто не держим много объектов в ней одновременно. 

• Первый байт у стрима, понятное дело, тоже получается быстрее. Буфер сидит и материализует весь список, прежде чем выдавить хоть байт. У стримов такой проблемы нет — как только что-то пришло, оно тут же отправляется.

• Пропускная способность у всех примерно одинаковая. Тут тоже понятно — она ограничена чуть другими условиями.

Хорошо, по замерам примерно разобрались. Стриминг во многом хорош. Давайте теперь рассмотрим его подводные камни.

Подводные камни стриминга

Важно учесть пару вещей, без которых «стрим» превращается обратно в буфер:

• Если пользуетесь EntityFramework, то нужно не забывать про AsNoTracking(). Иначе change tracker удержит каждую сущность, и набор всё равно осядет в памяти целиком.

• Контекст нужно брать из IDbContextFactory, а не как scoped. IAsyncEnumerable перечисляется уже после выхода из обработчика, и scoped-контекст успеет закрыться прямо посреди стрима.

• EnableRetryOnFailure буферит весь резалтсет — стратегия повторов не умеет переиграть наполовину прочитанный поток. Так что на стриминговых апи ее нужно отключать отдельно.

• И самое неочевидное — медленный консьюмер. Пока он чуть-по-чуть вычитывает ответ, сервер вынужден держать открытым соединение с базой всё это время (он же тянет строки по мере отдачи). С буфером у вас больше контроля — прочитали из базы, отпустили соединение обратно в ConnectionPool и дальше неспешно шлете все из памяти. Нужно иметь это в виду при реализации стрима.

А меняется ли что-то на HTTP/1.1, 2 и 3?

Когда я показывал ребятам на работе все вышеперечисленные наблюдения, то сразу же возник вопрос, а что будет на разных версиях HTTP? Будут ли работать все эти способы и что будет меняться?

Сразу отвечу на вопрос «будет ли работать?» — Да, все варианты будут работать на всех версиях. 

А что же будет отличаться?

Для нашего случая вся разница сводится к одному: сколько независимых потоков влезает в соединение и что происходит при потере пакета. 

HTTP/1.1 — одно TCP-соединение поддерживает один запрос-ответ за раз, но может открыть несколько соединений. Их браузер держит не больше ~6 на домен (лимит общий на все вкладки).

HTTP/2 — открывает одно TCP-соединение, но мультиплексирует несколько потоков данных, и все они последовательные, поэтому потеря пакета в одном потоке тормозит их все разом. 

HTTP/3 — тоже содержит мультиплексированные потоки данных, но поверх QUIC (UDP), и они независимы, поэтому потеря данных бьёт только по тому потоку, который эти данные потерял.

Всё остальное ниже — следствия из этих трёх строк. Это все применимо и к NDJSON, и к SSE.

• Конечное количество соединений на http1 не больше 6. Если на таком соединении открыть шесть стримов, то седьмой будет ждать своей очереди.

• Head-of-line blocking — только на HTTP/2. Один потерянный TCP-сегмент замораживает все стримы соединения сразу: TCP отдаёт байты строго по порядку, и пока клиент не дошлет потерянный сегмент, он держит в буфере всё, что пришло после него, — даже если те байты принадлежат другим стримам. То есть, если клиент запросил одновременно стрим с логами, картинку и статью — и при этом картинка залагала, — то это тормознет и прием ваших логов.

• Истощение пула соединений к базе на HTTP/2 и HTTP/3. Это довольно прикольный момент. На HTTP/1.1 клиент физически не сделает больше ~6 одновременных запросов к домену — и этот недостаток протокола ограничивал, сколько коннектов к базе клиент может занять разом. Это, по сути, спасало нас от истощения пула соединений. На HTTP/2 хватает одного соединения: в нём мультиплексируется до 100 потоков (у Kestrel по умолчанию MaxStreamsPerConnection = 100), а каждый медленный стрим держит коннект к БД на все время своей жизни. Получается, один клиент может занять до 100 соединений пула. HTTP/2 и HTTP/3 не то чтобы создают эту проблему, а скорее смещают бутылочное горлышко дальше: то есть, теперь мы упираемся в серверный лимит.

В целом для стриминга HTTP/3, конечно, предпочтителен по своей сути. Его 

киллер-фича для стриминга — отсутствие head-of-line blocking. Но есть небольшой момент: HTTP/3 включается не сразу. Браузер сначала идёт по HTTP/2, видит заголовок Alt-Svc и только следующим соединением поднимает HTTP/3. И если первый запрос вы сделали именно через SSE, то это соединение клиента может остаться на HTTP/2 до окончания сессии. В общем за этим нужно просто следить

И последнее, — что на практике ломает стрим чаще, чем выбор версии: промежуточные буферизирующие прокси. Nginx с дефолтным proxy_buffering on, корпоративные прокси, иные CDN — могут собрать весь ответ и отдать одним куском, то есть, делать ровно то, чего мы пытались избежать. Лечится proxy_buffering off / заголовком X-Accel-Buffering: no.

Итоги

В завершение хочу сказать, что стриминг иногда воспринимают как средство экономии памяти, но как мне кажется, это скорее просто более корректный вариант реализации определённых бизнес-сценариев: выгрузка всего одним проходом (экспорт, отчёт, ETL), живая лента, логи, прогресс долгой задачи, токены от LLM на лету. Общее у них одно — нет «страницы N», которую можно было бы запросить: данные либо нужны все за один проход, либо ещё рождаются в реальном времени.

А в остальном можно делать себе пагинацию и спокойно отдавать через ToListAsync.

А у вас стриминг где-то реально выручил — или, наоборот, выстрелил в ногу? Расскажите свой кейс в комментах, очень интересно почитать, где он зашёл, а где оказался лишним.

На этом все. Подписывайтесь на мой канальчик в Telegram C# Short Posts 18+, где я пишу о C#, инфре вокруг него и своих пет-проектах.

Ссылки

• Исходная статья: IAsyncEnumerable: Streaming Data Without Loading Everything Into

• Kestrel: Http2.MaxStreamsPerConnection

• nginx: ngx_http_proxy_module

• SSE behind nginx — баг с буферизацией (SO)

• InnoDB Row Formats (MySQL 8.0)

• Весь код примеров + k6-бенчмарки