errgroup в Go: параллельные горутины с обработкой ошибок

errgroup — это пакет golang.org/x/sync/errgroup, который решает задачу, где sync.WaitGroup уже не эффективен. WaitGroup справляется со сценарием «запустить N горутин и дождаться завершения», но при возврате ошибок из горутин его возможностей не хватает: нет ни сбора ошибок, ни отмены контекста при сбое, ни ограничения параллелизма. Всё это приходится реализовывать руками: каналы, мьютексы, обёртки. Пакет errgroup решает эту задачу одним специализированным типом с пятью методами.

Зачем нужен errgroup

Типичная ситуация: нужно параллельно запросить данные из нескольких источников и вернуть агрегированный результат. Реализация через sync.WaitGroup выглядит так:

func fetchAll(ctx context.Context, urls []string) ([]Response, error) {
    var (
        wg      sync.WaitGroup
        mu      sync.Mutex
        results []Response
        firstErr error
    )

    for _, url := range urls {
        wg.Add(1)
        go func() {
            defer wg.Done()

            resp, err := fetch(ctx, url)

            mu.Lock()
            defer mu.Unlock()
            if err != nil && firstErr == nil {
                firstErr = err
                return
            }
            results = append(results, resp)
        }()
    }

    wg.Wait()
    return results, firstErr
}

Мьютекс для защиты общего состояния, отдельная переменная для первой ошибки, ручной Add/Done. При ошибке в одной из горутин остальные продолжают работать, хотя результат будет отброшен из-за firstErr != nil т.к. частичные результаты игнорируются вызывающим кодом. И нет никакого способа ограничить параллелизм: если в urls тысяча элементов, то и горутин запустится тысяча одновременно.

Теперь та же логика, но уже через errgroup:

func fetchAll(ctx context.Context, urls []string) ([]Response, error) {
    g, ctx := errgroup.WithContext(ctx)

    var (
        mu      sync.Mutex
        results []Response
    )

    for _, url := range urls {
        g.Go(func() error {
            resp, err := fetch(ctx, url)
            if err != nil {
                return err
            }

            mu.Lock()
            results = append(results, resp)
            mu.Unlock()
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

Ошибки собираются автоматически. При первой ошибке контекст отменяется, и остальные горутины (если корректно обрабатывают ctx) понимают, что пора отменять работу. Никаких Add/Done потому что g.Go берёт это на себя.

API пакета

Пакет живёт в golang.org/x/sync/errgroup. Формально это не стандартная библиотека, но фактически «расширенная стандартная»: репозиторий golang.org/x/sync ведёт команда Go и держит те же стандарты качества.

$ go get golang.org/x/sync

Весь API это тип Group и пять методов.

Создание группы

Два способа создать группу:

// Без контекста: нулевое значение, как sync.WaitGroup
var g errgroup.Group

// С контекстом: отмена при первой ошибке
g, ctx := errgroup.WithContext(ctx)

Нулевое значение errgroup.Group валидно так же как и у sync.WaitGroup. Разница в том, что без WithContext контекст не отменяется при ошибке.

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

Go

func (g *Group) Go(f func() error)

Запускает функцию в новой горутине. Под капотом вызывает Add(1) перед запуском и Done() через defer и обеспечивает те же гарантии, что и у sync.WaitGroup.Go из Go 1.25, только с поддержкой ошибок.

Если установлен лимит через SetLimit и количество активных горутин уже на максимуме, Go блокируется до тех пор, пока одна из запущенных горутин не завершится.

Wait

func (g *Group) Wait() error

Блокируется до завершения всех горутин, запущенных через Go или TryGo. Возвращает первую ненулевую ошибку. Если все горутины завершились без ошибок, то возвращает nil.

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

SetLimit

func (g *Group) SetLimit(n int)

Ограничивает количество одновременно работающих горутин. -1 снимает ограничение. Важно: SetLimit нельзя вызывать, пока в группе уже есть активные горутины, т.к. это невалидное использование вызывающее панику.

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10) // Максимум 10 горутин одновременно

for _, url := range urls {
    g.Go(func() error {
        return fetch(ctx, url)
    })
}

Без SetLimit при тысяче URL запустится тысяча горутин. С SetLimit(10) одновременно работают максимум 10, остальные ждут своей очереди. По сути это семафор, встроенный прямо в группу без необходимости реализовывать отдельный канал-семафор или пул воркеров.

Внутри SetLimit реализован через буферизованный канал. При вызове Go токен записывается в канал до инкремента WaitGroup (если буфер заполнен то вызов блокируется), при завершении горутины токен извлекается из канала, освобождая место для следующей. Именно поэтому Go блокируется, а не просто стоит в очереди.

TryGo

func (g *Group) TryGo(f func() error) bool

Неблокирующая версия Go. При достижении лимита возвращает false вместо блокировки. Если есть свободный слот или лимит не установлен, то запустит горутину и вернёт true.

g.SetLimit(5)

for _, task := range tasks {
    if !g.TryGo(func() error {
        return process(task)
    }) {
        // Лимит достигнут, задача не запущена
        log.Printf("пропущена задача: %s (лимит достигнут)", task.ID)
    }
}

TryGo применяется, когда блокировка нежелательна. Например, в стриминговом обработчике с back-pressure: если консьюмеры не успевают обрабатывать события, вместо блокировки producer можно сбрасывать события с метрикой потерь. Или например цикл обработки событий, где задержка недопустима, но допустим пропуск задач. На практике нужен редко: обычно Go + SetLimit закрывают все потребности.

Контекст и отмена: WithContext в деталях

WithContext является основным отличием errgroup от простого «WaitGroup с ошибками». Давайте разберём, как это работает:

g, ctx := errgroup.WithContext(parentCtx)

Здесь ctx является дочерним контекстом от parentCtx. Её отмена происходит в двух случаях:

1. Любая горутина возвращает ненулевую ошибку
2. Wait возвращает управление (все горутины завершились)

При первой ошибке все остальные горутины получают сигнал через ctx.Done(). Если они проверяют контекст (а нормальный код это делает), то могут прекратить работу досрочно:

g, ctx := errgroup.WithContext(ctx)

// Горутина 1: завершится с ошибкой через 100ms
g.Go(func() error {
    time.Sleep(100 * time.Millisecond)
    return errors.New("сбой")
})

// Горутина 2: работает 10 секунд, но прервётся через ~100ms
g.Go(func() error {
    select {
    case <-time.After(10 * time.Second):
        return nil
    case <-ctx.Done():
        // Контекст отменён из-за ошибки в горутине 1
        return ctx.Err()
    }
})

err := g.Wait() // Вернёт ошибку "сбой" примерно через 100ms

Без WithContext вторая горутина проработала бы все 10 секунд, хотя результат уже не нужен.

Когда НЕ использовать WithContext

WithContext не серебряная пуля. Если горутины независимы и нужно дождаться завершения всех, даже если часть упала, берите нулевое значение Group:

var g errgroup.Group

// Закрываем несколько ресурсов параллельно.
// Каждый должен попытаться закрыться, даже если другой упал
g.Go(func() error { return db.Close() })
g.Go(func() error { return cache.Close() })
g.Go(func() error { return queue.Close() })

// Получим первую ошибку, но все три Close() точно отработают
err := g.Wait()

Тут WithContext только навредит: если db.Close() вернёт ошибку, контекст отменится, а cache.Close() и queue.Close() всё равно должны попытаться закрыть соединения. Потому что не каждая ошибка означает, что остальную работу нужно отменять.

Реальные сценарии

Параллельные HTTP-запросы с ограничением

Типичная задача: обогатить данные из нескольких API. Без ограничения параллелизма внешний сервис быстро начнёт отвечать 429 Too Many Requests:

func enrichUsers(ctx context.Context, users []User) ([]EnrichedUser, error) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(20) // Не больше 20 одновременных запросов

    results := make([]EnrichedUser, len(users))

    for i, user := range users {
        g.Go(func() error {
            profile, err := fetchProfile(ctx, user.ID)
            if err != nil {
                return fmt.Errorf("профиль пользователя %d: %w", user.ID, err)
            }
            results[i] = EnrichedUser{User: user, Profile: profile}
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

Заметьте: results[i] работает без мьютекса. Каждая горутина пишет строго в свой индекс, пересечений нет. Слайс заранее аллоцирован через make, горутины пишут в разные элементы, а запись в разные элементы слайса потокобезопасна.

Координация компонентов сервиса

errgroup отлично ложится на задачу запуска и координированной остановки компонентов приложения. Например, для HTTP-серверов с фоновыми воркерами (подробнее в статье про graceful shutdown):

func run(ctx context.Context) error {
    g, ctx := errgroup.WithContext(ctx)

    server := &http.Server{Addr: ":8080", Handler: newRouter()}

    g.Go(func() error {
        if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
            return err
        }
        return nil
    })

    g.Go(func() error {
        <-ctx.Done()
        shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()
        return server.Shutdown(shutdownCtx)
    })

    g.Go(func() error {
        return runWorker(ctx)
    })

    return g.Wait()
}

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

Параллельная обработка с записью в БД

Ещё одна частая задача: параллельно вычислить, последовательно записать. errgroup координирует горутины, а результаты стекаются через канал:

func processBatch(ctx context.Context, items []Item) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(10)

    results := make(chan Result, len(items))

    // Параллельная обработка
    for _, item := range items {
        g.Go(func() error {
            result, err := process(ctx, item)
            if err != nil {
                return fmt.Errorf("обработка %s: %w", item.ID, err)
            }
            results <- result
            return nil
        })
    }

    // Ждём завершения всех горутин и закрываем канал
    go func() {
        g.Wait()
        close(results)
    }()

    // Последовательная запись результатов.
    // Здесь нет deadlock, потому что запись выполняется в основной горутине,
    // а не через g.Go. Поэтому g.Wait() в закрывающей горутине корректно
    // дожидается всех producer'ов и закрывает канал.
    for result := range results {
        if err := save(ctx, result); err != nil {
            return fmt.Errorf("сохранение результата: %w", err)
        }
    }

    return g.Wait()
}

Повторный вызов g.Wait() безопасен, он просто вернёт тот же результат.

Паттерн pipeline: чтение — обработка — запись

Для pipeline-паттерна, где данные проходят через несколько стадий, errgroup координирует стадии, а каналы связывают их между собой:

func pipeline(ctx context.Context, input <-chan Raw) error {
    g, ctx := errgroup.WithContext(ctx)

    // Стадия 1: валидация
    validated := make(chan Validated, 100)
    g.Go(func() error {
        defer close(validated)
        for raw := range input {
            if err := ctx.Err(); err != nil {
                return err
            }
            v, err := validate(raw)
            if err != nil {
                return fmt.Errorf("валидация: %w", err)
            }
            validated <- v
        }
        return nil
    })

    // Стадия 2: обогащение (параллельно)
    enriched := make(chan Enriched, 100)

    // Используем отдельный WaitGroup для отслеживания завершения стадии 2,
    // чтобы избежать deadlock: g.Wait() в закрывающей горутине не должен
    // ждать завершения стадии 3, которая читает из enriched
    var stage2 sync.WaitGroup
    for i := 0; i < 5; i++ {
        stage2.Add(1)
        g.Go(func() error {
            defer stage2.Done()
            for v := range validated {
                if err := ctx.Err(); err != nil {
                    return err
                }
                e, err := enrich(ctx, v)
                if err != nil {
                    return fmt.Errorf("обогащение: %w", err)
                }
                enriched <- e
            }
            return nil
        })
    }

    // Закрываем enriched после завершения всех воркеров стадии 2
    go func() {
        stage2.Wait()
        close(enriched)
    }()

    // Стадия 3: запись
    g.Go(func() error {
        for e := range enriched {
            if err := save(ctx, e); err != nil {
                return fmt.Errorf("запись: %w", err)
            }
        }
        return nil
    })

    return g.Wait()
}

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

Сравнение с альтернативами

errgroup vs sync.WaitGroup

WaitGroup для горутин не возвращающих ошибки: параллельный запуск вычислений, fire-and-forget задачи. errgroup при необходимости обработки ошибок, а это подавляющее большинство реальных случаев.

errgroup vs каналы

Ручная координация через каналы — это более низкоуровневый подход. Он даёт больше контроля, но и требует большей сложности реализации:

func fetchAllWithChannels(ctx context.Context, urls []string) error {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    errs := make(chan error, len(urls))
    var wg sync.WaitGroup

    for _, url := range urls {
        wg.Add(1)
        go func() {
            defer wg.Done()
            if err := fetch(ctx, url); err != nil {
                errs <- err
                cancel() // Отменяем контекст при первой ошибке
            }
        }()
    }

    // Закрываем канал после завершения всех горутин
    go func() {
        wg.Wait()
        close(errs)
    }()

    // Собираем первую ошибку
    for err := range errs {
        return err
    }
    return nil
}

Результат тот же, а кода больше: канал для ошибок, WaitGroup для ожидания, ручной cancel, горутина для закрытия канала. В errgroup все эти детали спрятаны за простым API.

Каналы стоит брать, когда нужна нестандартная логика: сбор всех ошибок, частичные результаты, приоритеты между горутинами. А если задача сводится к «запустить, дождаться, вернуть ошибку», то всегда используйте errgroup.

errgroup vs пул воркеров

Самописный пул воркеров на каналах:

func workerPool(ctx context.Context, tasks []Task, workers int) error {
    taskCh := make(chan Task, len(tasks))
    errCh := make(chan error, 1)

    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for task := range taskCh {
                if err := task.Execute(ctx); err != nil {
                    select {
                    case errCh <- err:
                    default:
                    }
                    return
                }
            }
        }()
    }

    for _, task := range tasks {
        taskCh <- task
    }
    close(taskCh)

    wg.Wait()
    close(errCh)
    return <-errCh
}

С errgroup:

func workerPool(ctx context.Context, tasks []Task) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(10)

    for _, task := range tasks {
        g.Go(func() error {
            return task.Execute(ctx)
        })
    }

    return g.Wait()
}

SetLimit заменяет пул воркеров в большинстве случаев. Никаких каналов для задач, ручного управления воркерами или координации закрытия. Так код получается линейным и понятным с первого взгляда.

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

Ограничения и подводные камни

Только первая ошибка

Wait возвращает только первую ошибку. Если нужно видеть все:

func fetchAllCollectErrors(ctx context.Context, urls []string) error {
    // Используем нулевую группу без WithContext: не нужна отмена
    // контекста при ошибке, т.к. хотим собрать все ошибки
    var g errgroup.Group

    var (
        mu   sync.Mutex
        errs []error
    )

    for _, url := range urls {
        g.Go(func() error {
            // ctx прокидывается снаружи. Обратите внимание: при таком подходе,
            // если родительский контекст не будет отменён, все горутины отработают
            // до конца, даже если они уже бессмысленны из-за ошибок других.
            // Для задачи "собрать все ошибки" это нормально.
            if err := fetch(ctx, url); err != nil {
                mu.Lock()
                errs = append(errs, fmt.Errorf("%s: %w", url, err))
                mu.Unlock()
            }
            return nil // Возвращаем nil, чтобы errgroup не остановился
        })
    }

    g.Wait()
    return errors.Join(errs...)
}

Хитрость в том, что горутины возвращают nil, а ошибки складывают в слайс вручную. Зачем? Чтобы errgroup не отменил контекст при первой ошибке и дождался завершения всех горутин. В конце errors.Join склеивает все ошибки в одну.

Это вполне легитимное использование errgroup как «WaitGroup с лимитом и удобным API», просто без использования механики first-error. Координация горутин всё равно упрощается (не нужен отдельный WaitGroup), а сбор и обработку ошибок берём под ручной контроль.

Паника в горутине

Начиная с v0.14.0 (весна 2025) golang.org/x/sync/errgroup умеет перехватывать панику в горутине. Паника больше не убивает программу. Wait() перебрасывает её через тип errgroup.PanicValue (или errgroup.PanicError, если паника была значением error). Стектрейс при этом сохраняется:

var g errgroup.Group

g.Go(func() error {
    panic("что-то пошло не так")
})

// Wait() вызовет panic с типом errgroup.PanicValue,
// содержащим оригинальное значение и стектрейс
g.Wait()

При использовании WithContext паника в горутине тоже отменяет контекст группы, как и при возврате ненулевой ошибки. Остальные горутины узнают о проблеме и могут корректно завершиться.

Если вы хотите обработать панику как ошибку, а не перебрасывать, оборачивайте Wait в recover:

func safeWait(g *errgroup.Group) error {
    var err error
    func() {
        defer func() {
            if r := recover(); r != nil {
                // Работает только для v0.14.0+.
                // %v теряет стектрейс для errgroup.PanicValue
                // лучше извлечь r.Stack и залогировать отдельно
                err = fmt.Errorf("паника в горутине: %v", r)
            }
        }()
        err = g.Wait()
    }()
    return err
}

Важно: этот подход работает только для v0.14.0+. В более ранних версиях паника происходит внутри горутины, запущенной через g.Go, и этот recover её не поймает и программа упадёт. Для старых версий recover нужно было ставить внутри самой функции, передаваемой в Go.

Переиспользование группы

Формально errgroup.Group можно переиспользовать: после Wait запустить новые горутины через Go. Но не делайте так. Состояние ошибки от предыдущего запуска не очищается, и поведение может быть неожиданным. Проще создать новую группу для каждой порции задач.

Блокировка Go при SetLimit

Когда SetLimit установлен и все слоты заняты, Go блокируется. А если горутина, которая должна освободить слот, сама ждёт результата от горутины, которая не может запуститься, то получаем классический deadlock:

g.SetLimit(1)

g.Go(func() error {
    // Эта горутина заняла единственный слот.
    // g.Go ниже заблокируется навсегда, потому что слот не освободится,
    // пока текущая горутина не завершится.
    g.Go(func() error {
        return innerWork()
    })
    return nil
})

Deadlock возникает только при установленном SetLimit, то есть без лимита (или когда лимит заведомо больше числа вложенных задач) вложенный g.Go работает корректно. Правило: не вызывайте g.Go внутри горутины той же группы, если установлен SetLimit, то можно легко получить deadlock. Для вложенных задач создавайте отдельную группу.

Переменная цикла в замыкании

До Go 1.22 переменная цикла for переиспользовалась между итерациями. Знаменитый баг, когда все горутины получают последнее значение:

// Go < 1.22: будет баг и все горутины получат последний url
for _, url := range urls {
    g.Go(func() error {
        return fetch(ctx, url) // url одна и та же переменная
    })
}

// Исправление для Go < 1.22
for _, url := range urls {
    url := url // локальная копия
    g.Go(func() error {
        return fetch(ctx, url)
    })
}

С Go 1.22 переменная цикла создаётся заново на каждой итерации, и проблема ушла. Но если ваш проект ещё поддерживает Go 1.21 и ниже, то держите это в голове.

Не передавайте контекст errgroup в операции, которые должны завершиться

Если вы используете WithContext и контекст отменился, все операции на этом контексте тоже получат отмену. А это ловушка для cleanup-операций:

g, ctx := errgroup.WithContext(parentCtx)

g.Go(func() error {
    data, err := fetchData(ctx)
    if err != nil {
        return err
    }

    // Если другая горутина упала, ctx уже отменён.
    // saveData получит отменённый контекст и не сохранит данные.
    return saveData(ctx, data)
})

Для операций, которые обязаны завершиться несмотря ни на что, используйте родительский контекст или context.WithoutCancel (Go 1.21+):

return saveData(context.WithoutCancel(ctx), data)

Плюсы и минусы

Плюсы

• Меньше boilerplate: не нужно писать Add/Done, каналы для ошибок, мьютексы для сбора результатов
• Встроенный семафор: SetLimit заменяет самописный пул воркеров
• Отмена по контексту: WithContext автоматически сворачивает работу при первой ошибке
• Минималистичный API: пять методов покрывают большинство сценариев
• Качество стандартной библиотеки: пакет из golang.org/x, поддерживается командой Go, стабильный API

Минусы

• Внешняя зависимость: не стандартная библиотека, хотя и golang.org/x
• Только первая ошибка: для сбора всех ошибок нужен обходной путь
• Нет контроля над горутинами: нельзя отменить конкретную горутину, только весь контекст группы
• Нет приоритетов: все горутины равноправны, нет встроенного механизма приоритизации задач
• Ограниченный pipeline: для сложных pipeline-паттернов с несколькими стадиями и обратной связью errgroup может быть недостаточно

Когда что использовать

Заключение

errgroup занимает нишу между «голым» sync.WaitGroup и ручной координацией через каналы. Параллельные HTTP-запросы, batch-обработка, координация компонентов сервиса. Для этих задач его API более чем достаточно.

Перед использованием стоит ответить себе на три вопроса:

• Нужна ли отмена при первой ошибке? Если да, то используйте WithContext иначе нулевое значение.
• Можно ли запускать все горутины разом, или нужен SetLimit?
• Хватит ли знания о первой ошибке, или нужен полный отчёт?

От ответов зависит, какую конфигурацию errgroup выбрать.

FAQ

// sync.WaitGroup.Go функция без возвращаемого значения
wg.Go(func() {
    doWork()
})

// errgroup.Group.Go функция возвращает error
g.Go(func() error {
    return doWork()
})

results := make([]Result, len(items))
for i, item := range items {
    g.Go(func() error {
        r, err := process(item)
        if err != nil {
            return err
        }
        results[i] = r // Безопасно: каждая горутина пишет в свой индекс
        return nil
    })
}

var (
    mu      sync.Mutex
    results []Result
)
g.Go(func() error {
    r, err := process(item)
    if err != nil {
        return err
    }
    mu.Lock()
    results = append(results, r)
    mu.Unlock()
    return nil
})

panic: errgroup: modify limit while 1 goroutines in the group are still active

Теги:

• golang