Ограничение конкурентности в Go: errgroup, пулы воркеров и backpressure

Безграничный fan-out — это режим отказа по умолчанию в конкурентной среде Go: каждое выражение go добавляет ещё одну задачу в работу, без ограничения по количеству и без ответственного владельца. В этой статье речь о том, как задать и то, и другое.

Отмена и ограничение — это разные механизмы управления. Отмена определяет, когда работа прекращается; ограничение определяет, сколько задач существует одновременно. Сервис может идеально обрабатывать отмены и всё равно рухнуть, потому что десять тысяч горутин каждая по отдельности решили, что сейчас подходящий момент для обращения к базе данных.

Смежные с этой страницей материалы рассматривают другую половину проблемы. Go context.Context: правильное использование рассматривает context как контрольную плоскость для остановки работы; здесь же тема — плоскость пропускной способности: errgroup.SetLimit, каналы-семафоры, пулы воркеров и ограниченные очереди, а также одна форма утечки, которая переживает errgroup.Wait, даже когда все пути отмены выглядят подключёнными.

Пять примитивов, одна таблица решений

Два вопроса помогают выбрать строку таблицы. Первое: если одна задача завершилась ошибкой, нужны ли вам результаты остальных? Если да, то семантика fail-fast группы отбросит ту работу, которая вам нужна. Второе: приходит ли работа непрерывным потоком, или это фиксированная партия, которую можно посчитать перед запуском? Группы работают с партиями; пулы и очереди — с потоками.

Именно errors.Join делает обычную строку с WaitGroup жизнеспособной — он собирает ошибки всех воркеров, а не сохраняет только первую, что сочетается с правилами перевода границ, описанными в Архитектура обработки ошибок в Go.

Утечка, переживающая errgroup.Wait

Наиболее известная форма отказа errgroup не связана с ограничением. Это разрыв в отмене, который прячется за кодом, который выглядит корректно:

g, gctx := errgroup.WithContext(ctx)
results := make(chan int) // небуферизованный

go func() { // производитель — эту горутину некому владеть
    for i := 0; i < 100; i++ {
        results <- i
    }
    close(results)
}()

g.Go(func() error { // потребитель — член группы
    for r := range results {
        if r == 5 {
            return fmt.Errorf("save failed")
        }
    }
    return nil
})

err := g.Wait() // возвращает ошибку потребителя

Когда потребитель возвращается, gctx отменяется, и Wait возвращает управление. Производитель замер на results <- i — и отмена контекста не снимает блокировку отправки в канал. Поскольку никто не владеет этой горутин, она остаётся замороженной до конца жизненного цикла процесса. Если прогнать этот сценарий в цикле, утечка будет строго линейной:

// исправление: каждая операция с каналом внутри отменяемой задачи
// получает ветку Done
select {
case results <- i:
case <-gctx.Done():
    return
}

Такой же разрыв существует и на стороне приёма. Правило механическое: каждая операция с каналом внутри кода, который может быть отменён, должна иметь ветку <-ctx.Done(), и каждая горутина должна иметь владельца, который либо ждёт её, либо отменяет её. goleak в CI ловит случаи, которые пропущены при ревью, — он встраивается в тот же автоматизированный шлюз контроля качества, что и инструменты из Go Linters: необходимые инструменты для качества кода.

Два связанных сценария заслуживают внимания. Если производитель является членом группы, тот же код приводит не к утечке, а к взаимной блокировке — Wait блокируется навсегда, ожидая заблокированную отправку, что хотя бы падает громко. И производитель, который отправляет через select, но принимает из канала, чей потребитель уже завершился, утекает так же; ветки Done нужны на обеих сторонах.

errgroup.SetLimit: ограниченный fail-fast fan-out

SetLimit превращает errgroup в группу с контролем доступа. Каждый вызов Go, превышающий лимит, блокируется до освобождения слота:

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8)
for _, item := range items {
    g.Go(func() error {
        return process(ctx, item)
    })
}
err := g.Wait()

В производстве важны три семантики. Go блокируется, поэтому цикл выше применяет backpressure к производителю — обычно это именно то, что вам нужно, но это значит, что горутина вызывающей стороны может зависнуть, поэтому HTTP-обработчику, отправляющему большую партию, нужен собственный бюджет таймаута на весь цикл. SetLimit(0) блокирует каждый вызов Go навсегда. И изменение лимита, пока горутин активны, вызывает панику — лимит фиксируется на всё время жизни группы.

SetLimit наследует семантику группы: первая ошибка отменяет контекст, и Wait возвращает первую ошибку, отбрасывая результаты задач, которые ещё выполняются. Это правильный примитив, если отказ всей операции при первой ошибке допустим — получение одной страницы данных, вызов набора независимых health-check, рассылка запроса по шардам.

Каналы-семафоры: backpressure как фича

Буферизованный канал ёмкости n — это семафор без зависимостей:

sem := make(chan struct{}, 8)
var wg sync.WaitGroup
for _, item := range items {
    wg.Add(1)
    sem <- struct{}{} // блокируется, когда в работе 8 задач
    go func() {
        defer wg.Done()
        defer func() { <-sem }()
        _ = process(ctx, item)
    }()
}
wg.Wait()

Получение семафора происходит до оператора go, поэтому горутин создаются только тогда, когда слот существует. Размер буфера — это контракт: он одновременно является потолком конкурентности и очередью ожидающих разрешений.

Примитив оправдывает своё место, когда нужен контроль доступа, который не выражают группы: получение с учётом контекста, пулы на каждый downstream, или частичные результаты при ошибке:

select {
case sem <- struct{}{}: // слот получен
case <-ctx.Done():      // вызывающий сдался, не получив слот
    return ctx.Err()
}

При одинаковой нагрузке канал-семафор и SetLimit(8) завершаются в пределах 1,5% друг от друга — выбор между ними определяется семантикой, а не производительностью.

Пулы воркеров: для потоков, а не партий

Фиксированный пул долгоживущих воркеров разделяет приём (очередь) и выполнение (воркеры):

func pool(ctx context.Context, workers int) chan<- func() {
    jobs := make(chan func())
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case f, ok := <-jobs:
                    if !ok {
                        return
                    }
                    f()
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    return jobs
}

Пулы подходят для работы, которая приходит непрерывно — потребители очередей, циклы опроса, воркеры запросов, — там, где группа для каждой партии будет бесконечно создаваться и уничтожаться. Начиная с Go 1.25, sync.WaitGroup.Go убирает шаблонную пару Add/Done для распространённых случаев. Контракт завершения работы — это то, что нужно сделать правильно: закрытие jobs завершает воркеров после дренирования очереди; отмена ctx оставляет работающую в очереди. Оба варианта легитимны, но документированным поведением может быть только одно. Go 1.27 продолжает в том же направлении — дженерические методы, новые модернизаторы go fix и in-memory httptest.NewTestServer — всё это покрыто в Руководстве по апгрейду Go 1.27.

Ограниченные очереди: блокировать или отбрасывать

Буферизованный канал — это также очередь с потолком. Когда буфер заполняется, производитель блокируется — и этот момент является проектным решением. Блокировка передаёт давление вверх по цепочке тому, кто питает очередь; отбрасывание сбрасывает нагрузку ценой потери работы. Мониторинговый конвейер может отбрасывать; конвейер заказов должен блокироваться или сливаться в долговременное хранилище.

select {
case queue <- event: // принято
default:            // очередь заполнена: политика отбрасывания живёт здесь
    dropped.Add(1)
}

Какова бы ни была политика, делайте её явной и измеряемой. Молчаливая ветка default — это место, куда события уходят и исчезают.

Что стоит ограничение — измерения

Та же нагрузка, 10 000 задач с имитацией I/O по 5 мс каждая, Go 1.27.1:

Безграничный проигрывает по стенковому времени, потому что ничего в эксперименте не оказывает противодействие — десять тысяч таймерных горутин просто спят параллельно. Это и есть ловушка. Цена безграничного fan-out — никогда не CPU в вашем процессе; это десять тысяч одновременных соединений, downstream, который начинает таймаутиться под напором всплеска, и график памяти, который следует за глубиной очереди того, кому вы обращаетесь. На реальной зависимости безграничный проигрывает не в 18 мс — он таймаутится. Ограниченные проигрыши платят 6,6 секунд, потому что 10 000 задач ÷ 8 слотов × 5 мс — это 6,25 с неизбежной сериализации, и лимит — это суть: он превращает неконтролируемый всплеск в предсказуемый дренаж в 6,25 секунды.

Важно одно уточнение, когда лимит выше пары слотов: держите лимит на уровне или ниже того, что downstream способен поддерживать конкурентно, и выводите его из этого ограничения (размер пула соединений, rate limit, вместимость воркеров), а не из ощущения о «разумной» параллельности.

Измерение и тестирование ограниченного кода

Три сигнала покрывают большую часть. Тренды числа горутин (/sched/goroutines:goroutines, экспортируемые через Prometheus Go collector) ловят утечки как наклон. Латентность планировщика (/sched/latencies:seconds) ловит насыщенность — исполняемые горутин, ожидающие CPU, это реальное давление независимо от счётчика. Delta pprof из двух дампов горутин указывает, где живут замороженные горутин.

Для тестов ограниченные воркеры — тот случай, ради которого создавалось Тестирование конкурентного кода Go с synctest: фейковое время позволяет прогнать полный дренаж 10 000 задач за миллисекунды, synctest.Wait заменяет sleep-and-hope на достижение кватесенции, а пузырь быстро падает, если воркер надолго блокируется — воспроизведение утечки выше умирает в тестовом пузыре, а не в продакшене.

На уровне сервиса тот же вопрос об ограничении возникает между сервисами, где семафор становится rate limit, а circuit breaker защищает downstream — Паттерн Circuit Breaker в Go покрывает этот слой, а Go Microservices для оркестрации AI/ML показывает очереди, которые принимает оркестрационный слой.

Выбор

Таблица и диаграмма сжимаются в одну привычку: перед написанием go назовите потолок и владельца. Потолок — это SetLimit, семафор или ограниченная очередь; владелец — это Wait, ветка ctx.Done на каждой операции с каналом, или и то, и другое. Каждая точка создания горутин, на которую можно ответить на这两个 имени одним взглядом, — это та, которая не разбудит никого в 3 ночи.

Полезные ссылки

• Центр App Architecture — дизайн API, структура кода и паттерны интеграции
• Go context.Context: правильное использование — отмена, таймауты и значения