Очередь задач в Go: типичные ошибки при работе с воркерами

Задача: обработать поток фоновых заданий — рассылку писем, ресайз картинок, вызовы внешних API. Очевидное решение на Go — канал плюс пул горутин-воркеров. Оно пишется за пять минут и работает на демо. А потом в проде горутины растут до десятков тысяч, при деплое теряются наполовину обработанные задачи, а одна паника в воркере кладёт весь пул. Ниже — четыре ошибки, которые встречаются почти в каждой самодельной очереди, и что с ними делать. Всё актуально для Go 1.22+.

Ошибка 1. Горутины, которые никогда не завершаются

Классика: воркеры читают из канала, но никто не закрывает канал и не отменяет контекст. Продюсер перестал слать задачи — воркеры висят на <-jobs вечно. При каждом переподключении или пересоздании пула вы плодите новые горутины поверх старых.

// Плохо: воркер не знает, когда остановиться
func worker(jobs <-chan Job) {
    for job := range jobs {
        process(job)
    }
    // range завершится только при close(jobs),
    // а его часто забывают вызвать
}

Проблема в том, что for range по каналу выходит только когда канал закрыт. Если продюсер живёт всё время работы сервиса и канал не закрывается, воркеры не завершатся штатно даже при graceful shutdown. Правильный контроль — через context:

import (
    "context"
    "sync"
)

type Job struct {
    ID      int
    Payload []byte
}

func worker(ctx context.Context, jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-ctx.Done():
            // Контекст отменён — выходим, даже если в канале что-то есть
            return
        case job, ok := <-jobs:
            if !ok {
                return // Канал закрыт — задач больше не будет
            }
            process(ctx, job)
        }
    }
}

Как проверить, что горутины действительно уходят: перед shutdown и через пару секунд после него снимите runtime.NumGoroutine(). Число должно вернуться к базовому уровню, а не расти от рестарта к рестарту.

Утечка горутин не падает с ошибкой. Она проявляется как медленный рост памяти и числа горутин на графиках — и вы находите её не в логах, а в pprof через неделю после релиза.

Ошибка 2. Потеря задач при завершении

Второй частый промах связан с первым, но противоположен по последствиям. Вы аккуратно закрываете сервис, но по ctx.Done() воркеры выходят немедленно — вместе с задачами, которые уже лежат в канале, но ещё не взяты в обработку. При деплое каждый рестарт тихо теряет часть заданий.

Нужно разделить два сценария остановки:

• Мягкая — перестаём принимать новые задачи, но дорабатываем то, что уже в очереди. Продюсер закрывает канал, воркеры дочитывают его до конца через for range.
• Жёсткая — время вышло, бросаем всё. Отдельный контекст с таймаутом, по которому воркер прерывает даже текущую задачу.

// Продюсер при shutdown
func (p *Pool) Shutdown(timeout time.Duration) {
    close(p.jobs) // Новые задачи не принимаются, воркеры дочитают буфер

    done := make(chan struct{})
    go func() {
        p.wg.Wait()
        close(done)
    }()

    select {
    case <-done:
        // Все задачи из буфера обработаны
    case <-time.After(timeout):
        // Не успели за отведённое время — отменяем hard-контекст
        p.cancel()
    }
}

Важный нюанс: закрывать канал должна только та сторона, что в него пишет. Если продюсеров несколько, close из двух горутин даст панику close of closed channel. В этом случае координируйте закрытие через отдельную структуру или sync.Once.

Ошибка 3. Паника в воркере роняет весь пул

Одна задача с неожиданным входом — nil-указатель, деление на ноль, выход за границы среза — вызывает панику. Если она не перехвачена, горутина воркера умирает. Через сотню таких задач у вас нет ни одного живого воркера, а канал забит и блокирует продюсера.

Восстанавливать нужно на уровне обработки каждой задачи, а не всего цикла воркера — иначе после паники цикл всё равно прервётся:

func safeProcess(ctx context.Context, job Job) (err error) {
    defer func() {
        if r := recover(); r != nil {
            // Логируем, помечаем задачу как проваленную,
            // но воркер продолжает жить
            err = fmt.Errorf("паника при обработке задачи %d: %v", job.ID, r)
        }
    }()
    return process(ctx, job)
}

Вызов safeProcess ставится внутри цикла воркера. Тогда паника на одной задаче не выносит воркер: он логирует ошибку, при необходимости отправляет задачу в dead-letter и берёт следующую.

Отдельно предупреждение: recover ловит только панику той же горутины, в которой вызван defer. Если внутри process вы запускаете вложенную go func() и паника происходит там — этот recover её не поймает, и процесс всё равно упадёт. Вложенные горутины оборачивайте отдельно.

Ошибка 4. Неограниченное создание горутин вместо пула

Соблазн простой: на каждую входящую задачу — go process(job). Пока задач мало, работает. Под всплеском в 50 000 запросов вы получаете 50 000 горутин, каждая держит свой стек и, скорее всего, соединение с базой. Пул соединений исчерпан, память улетает, латентность растёт.

// Плохо: количество горутин ничем не ограничено
for job := range incoming {
    go process(job) // всплеск = взрыв
}

Фиксированный пул воркеров ограничивает параллелизм осознанно. Число воркеров подбирают под характер нагрузки, а не по числу ядер:

• CPU-bound (сжатие, хеширование) — примерно runtime.NumCPU(), больше только зря переключает контекст.
• IO-bound (запросы к БД, HTTP) — обычно больше числа ядер, но упирается в лимит пула соединений. Ставить воркеров больше, чем SetMaxOpenConns у пула БД, бессмысленно: лишние будут ждать соединение.

Сравнение подходов к параллелизму

Самодельная очередь в канале не переживает рестарт процесса: всё, что было в буфере, теряется. Если задачи нельзя потерять — это не про канал, а про брокер с персистентностью. Выбор между пулом и брокером — отдельная тема, здесь мы разбираем только ошибки внутрипроцессной очереди.

Что ломается на границах

• Буфер канала как «очередь». Большой буфер make(chan Job, 100000) маскирует проблему: продюсер не блокируется, задачи копятся в памяти, а при падении процесса исчезают все разом. Буфер — сглаживание всплесков, а не хранилище.
• Медленная задача блокирует воркер. Один воркер, зависший на 30-секундном HTTP без таймаута, — минус одна единица пропускной способности навсегда. Передавайте ctx во все внешние вызовы и ставьте им таймаут.
• WaitGroup.Add после старта горутин. Вызывайте wg.Add(n) до запуска воркеров, иначе Wait может вернуться раньше, чем горутины реально стартовали — редкая, но злая гонка.

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

Соберите базовую наблюдаемость до того, как что-то сломается в проде:

1. Метрика длины канала (len(jobs)) — если она стабильно у верхней границы, воркеров не хватает.
2. Метрика числа горутин — растёт от рестарта к рестарту, значит утечка.
3. Счётчик паник и проваленных задач — рост говорит о битых входных данных.
4. Профилирование через net/http/pprof — снимите goroutine-профиль под нагрузкой и после shutdown, сравните.

Когда потребуется персистентность или задачи должны переживать деплой, смотрите в сторону Redis (asynq, machinery) или NATS JetStream — но проверьте актуальные версии и API этих библиотек в их документации, они меняются.