Асинхронные операции

Почему долгая работа ломает синхронный ответ#

Посчитай на пальцах, что происходит с операцией на две минуты, оставленной в синхронном endpoint'е. Клиент держит соединение: его таймаут из урока 04 истечёт первым — ответ уйдёт в никуда, работа продолжится зря. Балансировщик посчитает инстанс живым, но занятым: соединения копятся, файловые дескрипторы тают, за пределом — ошибка accept. lb.Retry из урока 11 увидит таймаут и отправит вызов соседу — второй инстанс начнёт ту же двухминутную работу: ретраи, созданные для дребезга, умножают тяжёлую работу. Graceful shutdown из урока 19 будет ждать этой работы до своего бюджета — GracePeriod в конфигурации рассчитан на p99 в секундах, а не на минуты. Правило простое: если операция дольше нескольких секунд или результат нужен не немедленно, синхронный контракт не подходит — не потому, что HTTP так не умеет, а потому, что каждый механизм надёжности из курса работает против длинного ответа. Асинхронный контракт разрубает узел: быстрый ответ «принято», работа отдельно, результат по готовности.

Канон: задача вместо результата#

Канон stringsvc не ломается: Uppercase остаётся мгновенным, Count — дешёвым. Рядом появляется операция, честно названная долгой: UppercaseAll — верхним регистром целую пачку строк, с прогрессом и результатом, на который нельзя сослаться до готовности. Контракт асинхронного вызова — тройка договорённостей. Первая: ответ 202 Accepted — «работа принята, но не сделана»; не 200, потому что результата в ответе нет, и не 201, потому что нового ресурса пока тоже нет. Вторая: в ответе ID задачи и путь статуса — Location: /jobs/{id}, чтобы клиент ушёл опрашивать туда. Третья: Retry-After, когда очередь занята или когда хочется вежливо подсказать интервал опроса. Клиент, который не понял 202 и ждёт результат в теле, — признак плохого API-дока, а не плохого кода. И ещё одна честность: 202 — обещание «поставил», а не «сделаю»: ошибка исполнения всплывёт в статусе failed, и API статуса обязана уметь об этом сказать.

Endpoint, который принимает работу

// endpoint-слой из урока 17: тот же кит, другой контракт
func makeUppercaseAllEndpoint(svc StringService, queue *Queue) endpoint.Endpoint {
        return func(ctx context.Context, request interface{}) (interface{}, error) {
                req := request.(uppercaseAllRequest)
                if req.Input == "" {
                        return nil, ErrEmpty // канон ошибки из урока 03
                }
                job := newJob(req.Input)
                if err := queue.Put(ctx, job); err != nil {
                        return nil, err // ErrBusy уйдёт в 503 — блок ниже
                }
                return uppercaseAllResponse{
                        JobID:   job.ID,
                        Status:  "queued",
                        PollURL: "/jobs/" + job.ID,
                }, nil
        }
}

// transport-слой: 202 Accepted вместо 200
func encodeUppercaseAllResponse(ctx context.Context, w http.ResponseWriter, response interface{}) error {
        w.Header().Set("Location", response.(uppercaseAllResponse).PollURL)
        w.Header().Set("Retry-After", "2")
        w.WriteHeader(http.StatusAccepted) // 202 — работа принята
        return json.NewEncoder(w).Encode(response)
}

Endpoint-слой не изменился по форме: та же функция endpoint.Endpoint, тот же кит kithttp.NewServer из урока 04 — изменился контракт: вместо результата возвращается ссылка на будущее. Put — единственная дорога к очереди: endpoint не знает ни про канал, ни про воркеров, он знает про интерфейс Queue — это позволит в тестах из урока 16 подставить фейковую очередь. Порядок кода ответа задаётся в encode-функции: сначала заголовки, потом WriteHeader(202), потом тело — после WriteHeader заголовки уже не изменить.

Очередь в процессе: буферизованный канал#

Первая очередь, которую стоит построить, — уже в языке: make(chan *Job, capacity). Буферизованный канал — честная очередь с вместимостью: отправители кладут, пока есть свободные слоты, воркеры разбирают. Capacity — не магия, а поле Config из урока 18: он определяет, сколько всплеска готов сервис поглотить, и напрямую связан с памятью процесса — каждая задача в буфере живёт в куче целиком. Право отправки имеет только endpoint через Pool.Put — канал приватный, владелец один, и это убивает половину ошибок с гонками. Поиск «go-kit async» не выдаст очереди — асинхронность go kit не решается библиотекой сознательно: kit даёт endpoint'ы и middleware, исполнение задач — стандартный Go: канал, select, sync.WaitGroup. Такую конструкцию проще всего недооценить: in-memory очередь живёт в процессе и умирает с процессом — для задач, потерю которых пережить нельзя, дорога ведёт к внешнему брокеру, но до этого места — канала, пула и секвенции остановки — канал закрывает большинство потребностей.

Pool: воркеры, range и дренаж

type Pool struct {
        mu        sync.Mutex
        accepting bool
        jobs      chan *Job
        wg        sync.WaitGroup
        status    *StatusStore
}

func NewPool(workers, capacity int, status *StatusStore) *Pool {
        p := &Pool{
                accepting: true,
                jobs:      make(chan *Job, capacity),
                status:    status,
        }
        for i := 0; i < workers; i++ {
                p.wg.Add(1)
                go func() { // воркер живёт, пока канал не закрыт и буфер не пуст
                        defer p.wg.Done()
                        for job := range p.jobs {
                                p.status.Run(job) // queued -> running
                                p.status.Finish(job, run(job))
                        }
                }()
        }
        return p
}

Пул — это N одинаковых горутин над одним каналом, и больше ничего. Горутина-воркер одна на N задач, а не одна на запрос: ограничение параллелизма встроено в конструкцию, всплеск запросов не плодит горутины. range по каналу — готовый дренаж: когда канал закрыт, цикл добирает всё, что осталось в буфере, и только тогда выходит; то же поведение мы использовали в уроке 19 для батчера. defer p.wg.Done() первым делом горутины — счётчик уйдёт в ноль при любом выходе, включая панику; саму панику внутри run() всё равно перехватывает recover — warn-блок ниже.

Put и Close: очередь под защитой мьютекса

var ErrBusy = errors.New("queue full")
var ErrShuttingDown = errors.New("queue closed for new jobs")

func (p *Pool) Put(_ context.Context, j *Job) error {
        p.mu.Lock()
        defer p.mu.Unlock()
        if !p.accepting {
                return ErrShuttingDown // остановка началась — новых не берём
        }
        select {
        case p.jobs <- j:
                p.status.Queued(j)
                return nil
        default:
                return ErrBusy // очередь полна: backpressure, блок ниже
        }
}

func (p *Pool) Close() {
        p.mu.Lock()
        p.accepting = false
        close(p.jobs) // после последнего Put: воркеры доторкнут буфер
        p.mu.Unlock()
        p.wg.Wait() // заблокирует, пока последний воркер не выйдет
}

Мьютекс делает Put и Close атомарными относительно друг друга: либо Put успел отправить до close — и никакой паники, либо Close сработал первым — и Put видит accepting=false. Убери мьютекс — получишь классическую гонку «проверил флаг, а канал уже закрыт»: панику send on closed channel в лучшем запросе дня. Select с default — детектор переполнения: send в полный канал без default блокировался бы, превращая endpoint в точку затора; с default он мгновенно возвращает ErrBusy. Close повторяет секвенцию урока 19 в миниатюре: сначала запрет входа, потом close, потом wg.Wait — «новое не принимать, старое доделать».

Backpressure: что делать, когда очередь полна#

Очередь полна — это не авария, а честный ответ системы о её границе. Три стратегии, от честной к лживой. Отказ: Put вернул ErrBusy, endpoint транслирует его в 503 с Retry-After — клиент знает, что делать: подождать и повторить; код ошибки закрепи в codeFrom из урока 04, чтобы транспорт не превращал её в 500. Ожидание: send с таймаутом — запрос ждёт свободный слот ограниченное время; это сглаживает короткие всплески, но держит соединения — и работает только вместе с дедлайнами из урока 04, иначе снова копим висяки. Молчаливое отбрасывание: 202 при выброшенной задаче — худший вариант, обещание, которое не исполнят; допустимо только для фактов, чья потеря естественна (сэмплы, метрики), но не для пользовательских задач. Связи с курсом: rate limiter из урока 09 — первый шлагбаум, он отклоняет избыточный вход раньше очереди; circuit breaker из урока 10 защищает уже воркеров от падающей зависимости. Capacity из конфигурации — буфер на всплеск, а не склад бесконечности.

Статусы задач и идемпотентность#

Задача без статуса — контракта нет: клиент получил ID и теперь требует правду. StatusStore — map с RWMutex и четырьмя состояниями: queued (в буфере), running (воркер взял), done (результат готов), failed (ошибка с текстом). GET /jobs/{id} — обычный endpoint того же сервиса: makeStatusEndpoint, тот же kithttp.NewServer, тот же mux из урока 17; по готовности — результат или ссылка на него. Время жизни завершённых задач ограничено: джанитор раз в час вычищает старьё, иначе map станет утечкой памяти, которую найдут не сразу. Вторая половина контракта — идемпотентность принятия. Клиент не узнал, дошёл ли 202 (обрыв сети), повторил запрос — и получил вторую задачу с двойной работой. Лечится ключом идемпотентности: клиент присылает Idempotency-Key, endpoint ищет задачу с таким ключом и возвращает существующую вместо новой. Для повторных отправок 202 становится скучно безопасным: хоть сто раз — одна работа.

Фоновые задачи без запроса#

Не всякая фоновая задача отвечает чужому запросу: джанитор статусов из блока выше, прогрев кэша по расписанию, флаш батчей из урока 19, обновление статистики для /readyz из урока 20, ротация токенов. Механика одна и та же — и канонически минимальная: горутина-тикер с time.NewTicker, select из двух каналов — тик и остановка, — завершение по close с sync.WaitGroup. Три дисциплины. Одна горутина на одну периодическую задачу: не плодим планировщик-фреймворк там, где нужен один тикер. Паники перехватываем: падение тикера не должно ронять процесс — defer с recover и логом из урока 06. И работа тикера — тоже оценка для readiness: если джанитор перестал бежать, глубина map растёт, и это видно в метриках из урока 07 раньше, чем OOM. Если периодических задач становится много с разными расписаниями — вот тут первый раз оправдан cron-пакет вроде robfig/cron; но его горутины останавливаются тем же close, что и самодельный тикер.

Остановка: где дренаж пула в секвенции#

Пул встраивается в секвенцию урока 19 как ещё одна ступень между дренажом транспорта и закрытием зависимостей — и порядок не случаен. Пока srv.Shutdown дожидается активных запросов, клиентские 202 всё ещё законны: endpoint может класть задачи в очередь, значит, очередь закрывать рано. После дренажа HTTP новых входов нет — закрываем вход очереди: Pool.Close() ставит accepting=false, делает close(jobs), и воркеры добирают остаток буфера; wg.Wait() не пускает процесс дальше, пока последняя задача не найдёт свой статус. Только после этого — tracer.Close() и repo.Close(): задачи, доработавшие в дренаже, успеют записать результаты и спаны. Честное ограничение: если бюджет дренажа истёк, а буфер велик — недоработанные задачи пропадут, in-memory честно at-most-once; если потеря недопустима, интерфейс Queue из блока выше — то место, где in-memory реализация заменяется персистентной, а весь код endpoint'ов не меняется ни на строку.

Пул в секвенции остановки

case <-ctx.Done(): // сигнал из урока 19
        logger.Log("during", "shutdown", "msg", "signal received")

        registrar.Deregister() // 1. из ротации (урок 12)
        shutCtx, cancel := context.WithTimeout(context.Background(), cfg.GracePeriod)
        defer cancel()
        if err := srv.Shutdown(shutCtx); err != nil { // 2. дренаж HTTP
                logger.Log("during", "shutdown", "transport", "http", "err", err)
        }

        queue.Close() // 3. вход очереди закрыт, буфер доторкнут воркерами

        tracer.Close() // 4. спаны доработавших задач
        repo.Close()
        return nil

Ступень queue.Close() стоит между дренажом транспорта и телеметрией — и у этой позиции два объяснения. До HTTP-дренажа закрывать нельзя: активные запросы ещё ставят задачи, получишь либо панику, либо отказ на законном вызове. После tracer.Close() закрывать поздно: задачи, доторкавшие в дренаже, не оставят следов. Заметь, что таймаут GracePeriod теперь работает и на очередь: длинный буфер при коротком бюджете — дренаж оборвётся по DeadlineExceeded из srv.Shutdown, а queue.Close() всё равно добьёт остаток без таймаута; если это плохо, рядом с Close придумай бюджет и для воркеров.

Очередь полна, задача от клиента законна. Что должен вернуть endpoint?

Пул из 8 воркеров; в очереди 50 задач; параллельно пришло 30 запросов на постановку новых. Сколько горутин занято исполнением задач?

Что делают воркеры после close(jobs), если в буфере ещё 12 задач?

Гонка Put и Close

Прочитайте код и выберите, что он выведет:

// Вариант Put БЕЗ мьютекса — только проверка флага:
func (p *Pool) Put(_ context.Context, j *Job) error {
        if !p.accepting {
                return ErrShuttingDown
        }
        select {
        case p.jobs <- j:
                return nil
        default:
                return ErrBusy
        }
}

// Хронология двух горутин:
// HTTP-дренаж ещё идёт: горутина запроса прошла проверку accepting == true
// Пришёл SIGTERM, таймер истёк, pool.Close() успел:
// accepting = false; close(jobs)
// Запрос выполняет p.jobs <- j

Собери воркера пула

Восстанови четыре пропуска горутины-воркера: учёт WaitGroup, цикл с дренажем и переходы статусов.

p.wg.(1) go func() { defer p.wg.() for job := p.jobs { // выйдет после close и опустевшего буфера p.status.(job) // queued -> running p.status.Finish(job, run(job)) } }()

Путь асинхронной задачи

Расставь этапы от запроса до статуса done.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Синхронно против асинхронно#

Критерий	Синхронный endpoint	Асинхронный endpoint
HTTP-ответ	200 с результатом	202 с ID задачи и Location
Время ответа	равно времени работы	миллисекунды, работа идёт после
Таймаут клиента (урок 04)	риск для длинных операций	к самой работе не применим
Ретраи lb.Retry (урок 11)	безопасны для чтения	нужен ключ идемпотентности
Параллелизм	горутина на запрос	ровно N воркеров пула
Перегрузка	лимитер из урока 09 отклоняет вход	503 с Retry-After из очереди
Отказ операции	код ошибки в ответе	статус failed и текст в StatusStore
Наблюдаемость	трейс запроса (урок 08)	глубина очереди, возраст, статусы
Остановка	дренаж сервера (урок 19)	дренаж сервера плюс дренаж очереди

Не спавни горутину на каждый запрос

Искус искушений: отправил задачу — go func() { work() }() и вернул 202. Дёшево, пока не посчитаешь: горутина на запрос без потолка — всплеск в тысячу запросов рождает тысячу горутин с их стеками; память прощает, а планировщик — до поры. Хуже: паника в босой горутине валит весь процесс — recover по умолчанию никто не ставит, и ручное обслуживание 202 превращается в падение сервиса. Пул из этого урока закрывает обе проблемы: параллелизм ограничен N, а перехват паник — в одном месте.

Метрики очереди

Очередь без метрик — чёрный ящик с тайной жизнью. Три счётчика решают: глубина буфера (сколько задач ждёт), возраст старейшей задачи (когда очередь не разбирают), число ошибок ErrBusy (когда backpressure реально бил). Хочешь тоньше — гистограмма ожидания от постановки до взятия воркером: она покажет, где клиентское Retry-After врёт. Всё это — счётчики и гистограммы из урока 07, а пороговые алерты на глубину поставят тебя в известность раньше, чем первый клиент напишет в поддержку.

Честно о запуске кода

Очередь и пул — часть реального процесса, в песочнице курса они не исполняются, поэтому интерактивы построены на чтении кода. Локально всё оживёт: собери пул из урока, подави нагрузку генератором запросов и посмотри на глубину очереди, а затем Ctrl+C и прочитай по логам секвенцию дренажа до exit 0.

Термины урока

Термин

202 Accepted

нажмите, чтобы перевернуть

Определение

Ответ «работа принята, но не сделана»: ID задачи и Location для опроса статуса; не 200 — результата нет, и не 201 — ресурс ещё не появился.

1 / 6

Что дальше#

Сервис научился принимать работу вне запроса: 202, очередь, пул, статусы, дренаж — и у асинхронности остался последний неотвеченный вопрос: как о фактах узнать тем, кто не спрашивал? Клиент опрашивает /jobs/{id} — а аудитор, кэш и соседние сервисы тоже хотят знать, что UppercaseAll завершён, без поллинга и без связности кода. Урок 22-sobytiya введёт события: конверт факта, шину на каналах, middleware-издателя поверх канона урока 15 и честную таблицу гарантий доставки — от in-memory до NATS и Kafka. Карта курса — на странице курса; таймауты, против которых направлена асинхронность, — в уроке 04-http-transport; метрики очереди — в уроке 07-metriki; ретраи lb.Retry, множащие длинную работу, — в уроке 11-lb-proxy; секвенция остановки — в уроке 19-graceful.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Асинхронный endpoint — контракт: принять работу, вернуть 202 Accepted с ID и Location, исполнить отдельно.
Очередь в процессе — буферизованный канал с capacity из Config; владелец один, отправка через Pool.Put под мьютексом.
Worker pool — N горутин над одним каналом: параллелизм ограничен конструкцией; range даёт дренаж при close.
Backpressure — 503 с Retry-After по ErrBusy; select с default ловит переполнение без блокировки.
Статусы: queued/running/done/failed в StatusStore, GET /jobs/{id}; идемпотентность принятия — по Idempotency-Key.
Put и Close защищены одним мьютексом — иначе гонка с паникой send on closed channel.
Дренаж пула — accepting=false, close(jobs), wg.Wait(): ступень между дренажем HTTP и закрытием зависимостей.
In-memory очередь честно at-most-once: для неприемлемых потерь интерфейс Queue заменяется на персистентную реализацию.

Домашнее задание

Поставь асинхронность в stringsvc. Первое: добавь UppercaseAll — endpoint с 202, Location и Retry-After, очередь на канале с capacity из Config и пул из N воркеров; параметризуй N и capacity флагами из урока 18. Второе: собери StatusStore с четырьмя статусами и GET /jobs/{id}; проверь опросом, что задача проходит queued → running → done. Третье: добавь Idempotency-Key: повторная отправка с тем же ключом возвращает тот же jobID и не создаёт работу. Четвёртое: подави очередь (capacity 2, воркеры спят через time.Sleep в run) и убедись, что третий запрос получает 503 с Retry-After, а не висит. Пятое: запусти остановку с двадцатью задачами в буфере и GracePeriod из урока 19 — по логам секвенция должна показать дренаж HTTP, потом queue.Close() и wg.Wait, и все двадцать задач должны доторкнуть статусы; затем сравни с kill -9 — часть задач не получит done, и это честная цена in-memory.

Вопросы и ответы по уроку

Почему бы не запускать горутину прямо в endpoint'е и не городить пул?

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

Задачи теряются при падении процесса — зачем тогда вообще in-memory очередь?

Потеря — цена скорости и простоты: канал не пишет на диск, дренаж при штатной остановке спасает буфер, но kill -9 не спасёт ничто. Для задач, чья потеря терпима (предрасчёты, сэмплы, прогревы), это правильный инструмент. Для неприемлемых потерь интерфейс Queue из урока — шов: под ним подменяется персистентная реализация (брокер или таблица задач в БД), а endpoint'ы и статусы не меняются; переезд на брокер — это замена одного типа, а не ревизия сервиса.

Какой размер очереди и сколько воркеров выбирать?

Из измерений, а не из вкуса. Воркеры — исходя из природы работы: CPU-задачи — около числа ядер, IO-задачи — по пропускной способности зависимости; N задаётся флагом из урока 18 и проверяется под нагрузкой по p99. Capacity — буфер на длительность всплеска: глубина очереди на графике из урока 07 подскажет, сколько задач накапливается между пиками; буфер, который никогда не наполняется, — трата памяти, буфер, который всегда полон, — сигнал масштабировать воркеров или вводить лимитер из урока 09.

Термины урока

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

Почему долгая работа ломает синхронный ответ#

Канон: задача вместо результата#

Endpoint, который принимает работу

Очередь в процессе: буферизованный канал#

Pool: воркеры, range и дренаж

Put и Close: очередь под защитой мьютекса

Backpressure: что делать, когда очередь полна#

Статусы задач и идемпотентность#

Фоновые задачи без запроса#

Остановка: где дренаж пула в секвенции#

Пул в секвенции остановки

Гонка Put и Close

Собери воркера пула

Путь асинхронной задачи

Синхронно против асинхронно#

Термины урока

Что дальше#

Песочница урока

Коротко: что запомнить

Домашнее задание

Вопросы и ответы по уроку

Термины урока

Изучайте дальше

gRPC-Go: высокопроизводительный RPC