--- metadata: - name: generator content: Diplodoc Platform v5.50.6 alternate: - https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md - https://ytsaurus.tech/docs/ru/flow/cpp/examples/async_request.md --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt # Async Request в YTsaurus Flow (C++) [Пайплайн](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline) демонстрирует паттерн асинхронных внешних запросов с использованием [process functions](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md). События поступают на вход, преобразуются в запросы, обрабатываются детерминированной функцией и результаты накапливаются в [стейте](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state). [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/async_request) ## Компоненты пайплайна ### TStateKeeper `TStateKeeper` реализует `IProcessFunction` и запускается адаптером `TProcessFunctionComputation`. Функция использует `TSimpleExternalStateManager` для работы с внешним стейтом и обрабатывает два вида входных сообщений: - **Поток `event`**: при получении события создает `TRequestMessage` с уникальным `RequestId` и отправляет его в поток `request`. - **Поток `response`**: при получении ответа обновляет стейт — суммирует `total_length` из всех полученных ответов. Различение потоков происходит через `message->StreamId`. ### TRequestProcessor `TRequestProcessor` реализует `IProcessFunction` и запускается адаптером `TProcessFunctionSwiftMapComputation`, который не сохраняет входные и выходные сообщения в YTsaurus. Функция получает `TRequestMessage`, выполняет обработку (в данном примере — вычисляет длину запроса) и генерирует `TResponseMessage`. Использование SwiftMap-адаптера обосновано тем, что обработка запроса является чистой функцией: при одинаковых входных данных всегда генерируется одинаковый результат. ## Типы сообщений - **TEventMessage** — наследник `TYsonMessage`. Содержит поля `Key` и `Data`. - **TRequestMessage** — наследник `TYsonMessage`. Содержит поля `RequestId`, `Key` и `Request`. - **TResponseMessage** — наследник `TYsonMessage`. Содержит поля `RequestId`, `Key` и `Length`. Все типы сообщений регистрируются через макрос `YT_FLOW_DEFINE_YSON_MESSAGE`. ## Ключевой паттерн: цикл запрос-ответ Основная идея данного примера — построение цикла запрос-ответ внутри пайплайна с использованием нескольких потоков: 1. **events** → `TStateKeeper` → **request** (генерация запроса) 2. **request** → `TRequestProcessor` → **response** (обработка запроса) 3. **response** → `TStateKeeper` → стейт (накопление результатов) `TStateKeeper` одновременно является и потребителем событий, и потребителем ответов. Он использует `input_stream_ids = ["event", "response"]` и определяет тип входного сообщения по `StreamId`. В спеке process functions связываются с адаптерами явно: - для `state` указаны `computation_class_name = "NYT::NFlow::TProcessFunctionComputation"` и `processing_function = "NYT::NFlow::NExample::TStateKeeper"`; - для `processor` указаны `computation_class_name = "NYT::NFlow::TProcessFunctionSwiftMapComputation"` и `processing_function = "NYT::NFlow::NExample::TRequestProcessor"`. ## Управление стейтом `TStateKeeper` использует `TSimpleExternalStateManager` для хранения суммы длин всех обработанных запросов. Клиент стейта (`TMutableStateKeyClient`) привязывается в `Init(const IRuntimeInitContextPtr&)` через `initContext->InitExternalStateClient(StateClient_, "/state")`. Параметры стейта (`path` к таблице и т.п.) объявляются в секции `external_state_managers` [спеки](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec) компьютейшена. ## Функция main В `main` регистрируются три потока, а process functions регистрируются через `YT_FLOW_DEFINE_PROCESS_FUNCTION`: - `RegisterStream("event")` — входные события - `RegisterStream("request")` — запросы к процессору - `RegisterStream("response")` — ответы от процессора ## Исходный код ### TRequestProcessor {% code '/yt/yt/flow/examples/cpp/async_request/lib/async_request_functions.cpp' lang='cpp' lines='[BEGIN request_processor]-[END request_processor]' keep-indents %} ### TStateKeeper {% code '/yt/yt/flow/examples/cpp/async_request/lib/async_request_functions.cpp' lang='cpp' lines='[BEGIN state_keeper]-[END state_keeper]' keep-indents %} ## См. также - [Быстрый старт (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/getting-started.md) - [Computation (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/computation.md)