--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/go/examples/retryable_async_request.md - https://ytsaurus.tech/docs/ru/flow/go/examples/retryable_async_request.md - href: https://ytsaurus.tech/docs/ru/flow/go/examples/retryable_async_request.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Retryable Async Request в YTsaurus Flow (Go) Расширение примера [Async Request](https://ytsaurus.tech/docs/ru/flow/go/examples/async_request.md): обработчик запросов повторяет каждую неудачную попытку через [таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timer). Запрос целиком сохраняется во внутреннем [стейте](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state) и повторяется с фиксированной задержкой, пока внешний сервис на него не ответит. [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/retryable_async_request) ## Структура {#structure} Пайплайн состоит из трёх [компьютейшенов](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation): - `injector` — нативный сорс (`TSwiftPassthroughOrderedSourceComputation`), объявленный прямо в [спеке](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec): он читает очередь и публикует события в [стрим](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) `event`. Go-кода у него нет. - `state` (`stateKeeper`) — transform-компьютейшен, сгруппированный по `key`: на каждое событие он открывает запрос в стриме `request`, а пришедший в стриме `response` ответ прибавляет к суммарной длине ответов ключа во внешнем стейте `/state`. - `processor` (`requestProcessor`) — transform-компьютейшен, сгруппированный по `request_id`: он делает попытки обращения к внешнему сервису и публикует ответ в стрим `response`. Два компьютейшена сгруппированы по разным ключам не случайно: `processor` партиционируется по идентификатору запроса, поэтому запросы одного ключа повторяются независимо друг от друга. Таймерный стрим `delay` объявлен в спеке с `allow_timer_self_dependency = %true` — компьютейшен ставит таймеры сам себе. ## `main.go` {#main-go} Точка входа: создание пайплайна и регистрация обоих компьютейшенов компаньона. [Исходный код: `main.go`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/retryable_async_request/main.go) ```go type eventMessage struct { flow.YSONMessage Key uint64 `yson:"key"` Data string `yson:"data"` } type requestMessage struct { flow.YSONMessage RequestID uint64 `yson:"request_id"` Key uint64 `yson:"key"` Request string `yson:"request"` } type responseMessage struct { flow.YSONMessage RequestID uint64 `yson:"request_id"` Key uint64 `yson:"key"` Length int64 `yson:"length"` } func main() { pipeline := flow.NewPipeline() pipeline.AddStreams( flow.NewYSONStream[eventMessage](eventStreamID), flow.NewYSONStream[requestMessage](requestStreamID), flow.NewYSONStream[responseMessage](responseStreamID), ) pipeline.Add( flow.NewRowComputation("state", &stateKeeper{}), flow.NewRowComputation("processor", &requestProcessor{}), ) if err := pipeline.Run(); err != nil { fmt.Fprintf(os.Stderr, "retryable_async_request: %v\n", err) os.Exit(1) } } ``` ## `state_keeper.go` {#state-keeper-go} Маршрутизация стримов и накопление результатов во внешнем стейте: ветвление по `msg.StreamID` разводит событие и ответ по двум обработчикам. ```go type totalLengthState struct { TotalLength int64 `yson:"total_length"` } type stateKeeper struct{} var _ flow.RowFunction = (*stateKeeper)(nil) func (s *stateKeeper) OnMessage( ctx context.Context, rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector, ) error { switch msg.StreamID { case eventStreamID: return s.sendRequest(rt, msg, out) case responseStreamID: return s.bookResponse(rt, msg) default: return fmt.Errorf("unhandled stream %q", msg.StreamID) } } func (s *stateKeeper) sendRequest(rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector) error { var event eventMessage if err := msg.ConvertTo(&event); err != nil { return err } request := flow.NewYSONMessage[requestMessage](requestStreamID) request.RequestID = rand.Uint64() request.Key = event.Key request.Request = event.Data encoded, err := flow.ConvertFrom(rt, request) if err != nil { return err } out.AddMessage(encoded) return nil } func (s *stateKeeper) bookResponse(rt flow.Runtime, msg flow.ExtendedMessage) error { var response responseMessage if err := msg.ConvertTo(&response); err != nil { return err } state, err := flow.OpenExternalState(rt, totalStateName, msg) if err != nil { return err } var total totalLengthState if _, err := state.ConvertTo(&total); err != nil { return err } total.TotalLength += response.Length return state.ConvertFrom(&total) } ``` ## `request_processor.go` {#request-processor-go} Значение стейта — обычная Go-структура с YSON-тегами. В ней лежит всё, что нужно повтору, включая счётчик неудачных попыток, поэтому исходное сообщение повтору уже не требуется: ```go type requestState struct { RequestID uint64 `yson:"request_id"` Key uint64 `yson:"key"` Request string `yson:"request"` FailedAttempts int64 `yson:"failed_attempts"` } ``` Логика повторов: `OnMessage` открывает запрос и делает первую попытку, `OnTimer` читает запрос из стейта и повторяет её. Общий для обоих обработчиков метод `attempt` либо ставит таймер на следующую попытку, либо публикует ответ и очищает стейт: ```go type requestProcessor struct{} var ( _ flow.RowFunction = (*requestProcessor)(nil) _ flow.RowTimerFunction = (*requestProcessor)(nil) ) func (p *requestProcessor) OnMessage( ctx context.Context, rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector, ) error { state, err := flow.OpenYSONState[requestState](rt, requestStateName, msg) if err != nil { return err } var input requestMessage if err := msg.ConvertTo(&input); err != nil { return err } request := requestState{ RequestID: input.RequestID, Key: input.Key, Request: input.Request, } return p.attempt(rt, state, request, out) } func (p *requestProcessor) OnTimer( ctx context.Context, rt flow.Runtime, timer flow.Timer, out flow.OutputCollector, ) error { state, err := flow.OpenYSONState[requestState](rt, requestStateName, timer) if err != nil { return err } if state.Empty() { return nil } return p.attempt(rt, state, *state.Value(), out) } func (p *requestProcessor) attempt( rt flow.Runtime, state *flow.YSONState[requestState], request requestState, out flow.OutputCollector, ) error { if !succeeds(request) { request.FailedAttempts++ *state.Value() = request out.AddTimer(flow.TimerRequest{TriggerTimestamp: uint64(time.Now().Add(retryDelay).Unix())}) return nil } response := flow.NewYSONMessage[responseMessage](responseStreamID) response.RequestID = request.RequestID response.Key = request.Key response.Length = int64(len(request.Request)) encoded, err := flow.ConvertFrom(rt, response) if err != nil { return err } out.AddMessage(encoded) state.Clear() return nil } func succeeds(request requestState) bool { return (request.RequestID+uint64(request.FailedAttempts))%attemptsToSucceed == 0 } ``` ## Ключевые паттерны {#key-patterns} - Повторы через таймеры: `out.AddTimer(flow.TimerRequest{TriggerTimestamp: ...})` откладывает следующую попытку на `retryDelay`, а `OnTimer` её выполняет. Это стандартный способ реализовать повторы в Flow без внешних очередей. Пустой `StreamID` в `flow.TimerRequest` означает единственный таймерный стрим компьютейшена. - Счётчик попыток живёт в стейте вместе с данными запроса, поэтому переживает рестарт [воркера](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#worker): после перезапуска повтор продолжается с того же места. - Вход один раз преобразуется в `requestMessage` через `msg.ConvertTo(&input)`, а retry-логика работает с отдельной структурой `requestState`, содержащей счётчик неудачных попыток. - Один и тот же стейт открывается и по сообщению, и по таймеру: `flow.OpenYSONState[requestState](rt, requestStateName, msg)` и `flow.OpenYSONState[requestState](rt, requestStateName, timer)` — оба входа несут ключ группировки, по которому стейт и адресуется. - Очистка стейта при успехе: после `state.Clear()` сработавший позже таймер видит `state.Empty()` и ничего не делает — устаревшее срабатывание безвредно. - Разделение ответственности: `stateKeeper` ничего не знает о повторах, вся логика инкапсулирована в `requestProcessor`, поэтому стратегию повторов можно поменять, не трогая учёт результатов. - Детерминированная симуляция сбоев: `succeeds(request)` стоит на месте настоящего клиента внешнего сервиса — в реальном коде здесь был бы HTTP-вызов. ## См. также - [Быстрый старт (Go)](https://ytsaurus.tech/docs/ru/flow/go/getting-started.md) - [Async Request (Go)](https://ytsaurus.tech/docs/ru/flow/go/examples/async_request.md) - [Работа со стейтами (Go)](https://ytsaurus.tech/docs/ru/flow/go/state.md) - [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md) - [Аналогичный пример на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/retryable_async_request.md)