Пул воркеров в Go: рабочие приёмы без утечек горутин

Задача выглядит просто: обработать 50 000 ссылок, для каждой сделать HTTP-запрос и записать результат. Наивное решение — go process(url) в цикле. Через секунду у вас 50 000 горутин, все они одновременно долбят внешний API, тот отвечает 429, а память процесса вырастает на несколько сотен мегабайт под стеки горутин и буферы соединений. Параллелизм нужен ограниченный и управляемый — это и есть пул воркеров.

Разберём три рабочих приёма для Go 1.22+: пул на каналах, ограничитель на семафоре и errgroup с лимитом. У каждого своя цена — по коду, по контролю над ошибками и по отмене.

Почему «горутина на задачу» ломается

Горутина в Go дешёвая, но не бесплатная: стартовый стек 8 КБ, плюс каждая может держать открытое соединение, буфер, ссылку на большой объект. Проблема не в стоимости одной горутины, а в том, что вы теряете контроль над их числом. Симптомы под нагрузкой:

• внешний сервис возвращает 429 Too Many Requests или рвёт соединения, потому что вы открыли тысячи одновременно;
• потребление памяти растёт линейно с числом задач, а не остаётся константным;
• при ошибке или отмене контекста горутины продолжают работать — их некому остановить.

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

Приём 1. Пул на каналах

Классика: канал задач, канал результатов, фиксированное число воркеров. Читаем из jobs, пишем в results. sync.WaitGroup ждёт завершения всех воркеров.

package main

import (
	"fmt"
	"sync"
)

// Задача и результат — конкретные типы, а не interface{}
type job struct {
	id  int
	url string
}

type result struct {
	id    int
	value int
	err   error
}

func worker(wg *sync.WaitGroup, jobs <-chan job, results chan<- result) {
	defer wg.Done()
	// Воркер живёт, пока jobs открыт и в нём есть данные
	for j := range jobs {
		results <- result{id: j.id, value: len(j.url)}
	}
}

func main() {
	const workers = 8
	jobs := make(chan job, 100)
	results := make(chan result, 100)

	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go worker(&wg, jobs, results)
	}

	// Отдельная горутина закрывает results после того,
	// как все воркеры отработали. Без этого range по results зависнет
	go func() {
		wg.Wait()
		close(results)
	}()

	// Наполняем очередь и закрываем — это сигнал воркерам завершиться
	go func() {
		for i := 0; i < 1000; i++ {
			jobs <- job{id: i, url: fmt.Sprintf("url-%d", i)}
		}
		close(jobs)
	}()

	for r := range results {
		if r.err != nil {
			fmt.Printf("job %d failed: %v\n", r.id, r.err)
		}
	}
}

Три места, где здесь ошибаются:

• Забыли close(jobs) — воркеры навсегда застрянут в range jobs, а wg.Wait() не дождётся.
• Закрыли results из воркера — при нескольких воркерах получите панику «close of closed channel». Закрывать должен один, после wg.Wait().
• Читают results в той же горутине, что пишет jobs, при небуферизованных каналах — дедлок. Разводите чтение и запись по разным горутинам.

Приём 2. Семафор на буферизованном канале

Когда результат складывать в канал не нужно (пишете сразу в базу или в лог), а нужно лишь ограничить одновременность — семафор проще. Буферизованный канал ёмкостью N работает как счётчик слотов.

package main

import (
	"context"
	"sync"
)

func processAll(ctx context.Context, urls []string) {
	const limit = 10
	sem := make(chan struct{}, limit) // 10 слотов
	var wg sync.WaitGroup

	for _, u := range urls {
		select {
		case <-ctx.Done():
			// Контекст отменён — не запускаем новые задачи
			wg.Wait()
			return
		case sem <- struct{}{}: // занимаем слот, блокируемся если все заняты
		}

		wg.Add(1)
		go func(url string) {
			defer wg.Done()
			defer func() { <-sem }() // освобождаем слот
			processOne(ctx, url)
		}(u)
	}
	wg.Wait()
}

func processOne(ctx context.Context, url string) {
	// здесь ваш HTTP-запрос с ctx
}

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

Слот в семафоре и живой воркер — не одно и то же. Семафор ограничивает, сколько задач бегут сейчас. Пул задаёт, сколько исполнителей существует. Под ровным потоком выбирайте пул, под всплесками — семафор.

Приём 3. errgroup с лимитом

Если задачи возвращают ошибку и при первой же вы хотите отменить остальные — golang.org/x/sync/errgroup закрывает это без ручного управления каналами. Метод SetLimit ограничивает число одновременных горутин.

package main

import (
	"context"

	"golang.org/x/sync/errgroup"
)

func fetchAll(ctx context.Context, urls []string) error {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(10) // не больше 10 горутин одновременно

	for _, u := range urls {
		url := u // до Go 1.22 обязательно, с 1.22 переменная и так своя
		g.Go(func() error {
			return fetchOne(ctx, url)
		})
	}

	// Wait вернёт первую ненулевую ошибку и отменит общий ctx
	return g.Wait()
}

func fetchOne(ctx context.Context, url string) error {
	return nil
}

errgroup.WithContext даёт производный контекст, который отменяется при первой ошибке любой горутины. Внутри fetchOne проверяйте ctx.Err() или передавайте ctx в http.NewRequestWithContext — иначе отмена ни на что не повлияет и «висящие» задачи доработают до конца.

Когда что выбирать

Что ломается на практике

Типичные ошибки, которые проходят тесты и всплывают в проде:

• Утечка горутин. Воркер читает из канала без ctx и без close. Если продюсер упал, а канал не закрыт, воркеры висят вечно. Проверяйте через runtime.NumGoroutine() в метриках: после завершения пакета число должно вернуться к базовому.
• Игнор контекста внутри задачи. SetLimit и семафор ограничивают старт, но не прерывают уже начатую работу. Долгий HTTP-запрос без ctx доработает даже после отмены. Всегда прокидывайте ctx в сетевые вызовы.
• Захват переменной цикла. До Go 1.22 go func(){ use(u) }() без явного копирования u ловил последнее значение. С Go 1.22 переменная цикла своя на каждой итерации — но если поддерживаете старые версии, копирование обязательно.
• Небуферизованный канал результатов при однопоточном чтении. Воркеры пишут в results, читатель ещё не начал — все воркеры блокируются, пул стоит. Дайте каналу буфер или запускайте читателя раньше продюсера.

Что мерить и что настроить дальше

Подберите число воркеров под характер задач, а не наугад. Для CPU-bound работы отталкивайтесь от runtime.GOMAXPROCS(0); для IO-bound (сеть, диск) оптимум обычно выше числа ядер, потому что горутины большую часть времени ждут. Мерьте под реальной нагрузкой: go test -bench с разными лимитами и pprof по горутинам и памяти покажут, где полка.

Дальше смотрите в сторону динамического бэкпрешера — если очередь jobs переполняется быстрее, чем воркеры разбирают, буферизованный канал начнёт блокировать продюсера, и это нормальный сигнал притормозить приём. Инвалидация и повторы упавших задач — отдельная тема, здесь мы её не разворачиваем.