Сейчас неконтролируемые вызовы go func() встречаются уже не так часто, как на заре Go.

Наверняка многие знают о принципе: «Никогда не запускайте горутину, если не знаете, как она завершится». И всё же мне до сих пор время от времени попадаются несинхронизированные вызовы go func(), которые запускают задачи без ожидания результата.

Я не говорю о заданиях, переданных в специализированную очередь задач вроде Asynq. В таком случае система управления задачами сама отвечает за жизненный цикл запущенных задач. Речь о небольших best‑effort‑задачах, для которых особенно велик соблазн просто вызвать go func() и забыть о них. Например, так часто отправляют уведомления или записывают ресурсоёмкие диагностические логи.

Обычно это выглядит так:

func (h *handler) createOrder(w http.ResponseWriter, r *http.Request) {
    user := r.URL.Query().Get("user")
    ctx := r.Context()

    go func() {
        h.tasks.SendNotification(ctx, user) // (1)
    }()
    go func() {
        h.tasks.WriteDiagnosticLog(ctx, user) // (2)

    }()

    w.WriteHeader(http.StatusAccepted)
}

В обработчике выше:

1. В фоновой горутине отправляется уведомление.

2. В другой фоновой горутине записывается диагностический лог.

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

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

• обе задачи получают r.Context(), а net/http отменяет его после завершения обработчика. Если задачи учитывают отмену — а в идеале так и должно быть, — они могут завершиться, не успев выполнить работу до конца;

• паника в любой из горутин завершит весь процесс, поскольку её никто не перехватывает;

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

Все задачи проходят через один пул воркеров

Один из вариантов решения, которым я пользуюсь, — небольшой пул воркеров на основе буферизованного канала:

• каждая фоновая задача отправляется в канал в виде замыкания func();

• воркеры запускаются в точке сборки приложения, обычно в main, вычитывают задачи из канала и выполняют замыкания;

• количество воркеров фиксировано, поэтому уровень параллелизма остаётся ограниченным;

• никакой другой код в приложении не запускает фоновые горутины: вся фоновая работа проходит через этот пул;

• при завершении работы main закрывает канал.

В простейшем виде это выглядит так:

// Создаём буферизованный канал в main.
tasks := make(chan func(), 64)

// Там же, в main, запускаем фиксированное число воркеров,
// которые вычитывают задачи из канала и выполняют их.
for range 4 {
   go func() {
       for task := range tasks {
           task()
       }
   }()
}

// Из обработчика или любого другого места
// отправляем задачу в виде замыкания.
tasks <- func() {
   doWork()
}

// При завершении работы закрываем канал.
close(tasks)

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

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

Для настоящего пула одного канала недостаточно

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

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

• передача контекста: задача не должна наследовать отмену контекста запроса, но значения из него ей всё ещё могут понадобиться;

• тайм‑ауты: после отвязки от контекста запроса задачу больше ничто не отменяет, поэтому ей нужен собственный дедлайн;

• корректное завершение работы: после сигнала завершения все принятые задачи должны успеть выполниться до выхода из main.

Следующий API призван закрыть все эти требования.

Весь API состоит из трёх функций

Полная реализация — это небольшой пакет bg, который оборачивает канал в структуру Background:

// internal/bg/bg.go
type Background struct {
    tasks   chan func()
    workers sync.WaitGroup
    onPanic func(any)

    mu      sync.RWMutex
    stopped bool
}

func New(capacity, workers int, onPanic func(any)) *Background
func (b *Background) Submit(task func()) bool
func (b *Background) Stop()

New создаёт пул, Submit ставит задачу в очередь, а Stop завершает работу пула. Конструктор создаёт канал и запускает воркеры — проверка аргументов здесь опущена:

// internal/bg/bg.go
func New(capacity, workers int, onPanic func(any)) *Background {
    background := &Background{
        tasks:   make(chan func(), capacity),
        onPanic: onPanic,
    }
    for range workers {
        background.workers.Go(background.work)
    }
    return background
}

workers.Go — это метод WaitGroup.Go, добавленный в Go 1.25. Он одним вызовом запускает горутину каждого воркера и добавляет её в группу. Во время завершения работы Stop ждёт завершения этой группы.

Каждый воркер вычитывает задачи из канала:

// internal/bg/bg.go
func (b *Background) work() {
    for task := range b.tasks {
        b.run(task)
    }
}

Это тот же цикл чтения из канала, что и в предыдущем наброске. Он продолжает получать задачи, пока канал не будет закрыт и полностью вычитан. Аргументы capacity и workers ограничивают параллелизм: одновременно выполняется не более workers задач, а ещё не более capacity могут ожидать в очереди.

Submit отправляет задачу в этот канал:

// internal/bg/bg.go
func (b *Background) Submit(task func()) bool {
    b.mu.RLock()
    defer b.mu.RUnlock()

    if b.stopped {
        return false
    }
    b.tasks <- task
    return true
}

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

sync.RWMutex нужен потому, что отправка в закрытый канал вызывает панику. Stop закрывает канал, поэтому пул должен гарантировать, что в этот момент ни один вызов Submit не находится посреди отправки.

Во время отправки Submit удерживает блокировку на чтение, которую могут одновременно разделять сколько угодно горутин. Stop закрывает канал, удерживая блокировку на запись, а получить её можно только после того, как все читатели освободят свои блокировки. Поэтому закрытие канала никогда не произойдёт посреди отправки. А после выполнения Stop последующий вызов Submit увидит stopped и вернёт false, не пытаясь ничего отправить.

Обычный sync.Mutex тоже защитил бы от паники, но без всякой пользы заставил бы все вызовы Submit выполняться строго по одному. Отправлять данные в канал из нескольких горутин и так безопасно, поэтому защищать отправителей друг от друга не нужно. Нужно лишь отделить закрытие канала от отправок — и две стороны RWMutex решают именно эту задачу.

Принятые задачи воркер запускает через run:

// internal/bg/bg.go
func (b *Background) run(task func()) {
    defer func() {
        if value := recover(); value != nil {
            b.handlePanic(value)
        }
    }()
    task()
}

func (b *Background) handlePanic(value any) {
    if b.onPanic == nil {
        return
    }
    defer func() {
        _ = recover()
    }()
    b.onPanic(value)
}

run оборачивает каждую задачу в отложенный вызов recover. Если задача вызывает панику, воркер перехватывает её значение и не даёт ей завершить весь процесс. Затем он передаёт это значение в handlePanic и переходит к следующей задаче.

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

Так решается часть обработки ошибок, связанная с паниками. Обычные ошибки остаются внутри задачи — к ним мы вернёмся в следующем разделе.

Stop прекращает приём новых задач и ждёт завершения воркеров:

// internal/bg/bg.go
func (b *Background) Stop() {
    b.mu.Lock()
    if !b.stopped {
        b.stopped = true
        close(b.tasks)
    }
    b.mu.Unlock()

    b.workers.Wait()
}

Stop захватывает блокировку на запись, устанавливает stopped, чтобы Submit больше не принимал новые задачи, и закрывает канал. Закрытие канала не удаляет значения, которые уже находятся в его буфере. Воркеры завершают обработку всех принятых задач, после чего выходят из циклов чтения. workers.Wait возвращает управление, когда завершатся все воркеры.

Stop можно безопасно вызывать несколько раз, в том числе из разных горутин. Это та часть корректного завершения работы, за которую отвечает пул. Вторая часть — порядок вызовов в main; к нему мы вернёмся в конце.

Задачи создают собственные контексты

Сигнатура задачи — просто func(). Она не принимает контекст и не возвращает ошибку.

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

// order/handler.go
user := r.URL.Query().Get("user")
detached := context.WithoutCancel(r.Context()) // (1)

h.background.Submit(func() {
    ctx, cancel := context.WithTimeout(
        context.Background(), // (2)
        h.taskTimeout,
    )
    defer cancel()
    h.tasks.SendNotification(ctx, user)
})
h.background.Submit(func() {
    ctx, cancel := context.WithTimeout(
        detached, // (3)
        h.taskTimeout,
    )
    defer cancel()
    h.tasks.WriteDiagnosticLog(ctx, user)
})

В этих двух вызовах Submit:

1. context.WithoutCancel сохраняет значения из контекста запроса, например идентификатор запроса, но убирает его отмену и дедлайн;

2. для отправки уведомления достаточно уже извлечённого значения user, поэтому контекст создаётся с нуля;

3. диагностическому логу могут понадобиться и значения из контекста запроса, поэтому его контекст строится поверх отвязанного.

Оба тайм‑аута начинают отсчитываться в момент, когда воркер вызывает замыкание, а не пока задача ждёт в очереди. Методы задач должны передавать контекст во все блокирующие вызовы. Тайм‑аут не сможет остановить код, который игнорирует контекст.

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

Сначала остановите сервер, затем пул

Основная программа создаёт пул до запуска HTTP‑сервера:

// cmd/main.go
background := bg.New(64, 4, logPanic)
defer background.Stop()

В этом примере в очереди могут ожидать до 64 задач, четыре воркера обрабатывают её содержимое, а logPanic записывает в лог значение паники и стек вызовов. Благодаря defer пул корректно завершает работу при любом выходе из main.

Получив сигнал завершения, main сначала останавливает HTTP‑сервер, а уже затем воркеры:

// cmd/main.go
shutdownCtx, cancel := context.WithTimeout(
    context.WithoutCancel(ctx),
    2*time.Second,
)
defer cancel()

if err := server.Shutdown(shutdownCtx); err != nil {
    return err
}

// Отложенный background.Stop() выполнится при выходе из main.

К этому моменту сигнал уже отменил ctx, поэтому WithoutCancel убирает эту отмену, после чего WithTimeout добавляет дедлайн в две секунды.

Server.Shutdown прекращает приём новых запросов и ждёт завершения активных обработчиков. Когда он успешно возвращает управление, все эти обработчики уже завершили свои вызовы Submit. Затем отложенный Background.Stop прекращает приём новых задач, завершает обработку всех принятых задач и ждёт остановки воркеров.

Это по‑прежнему очередь в памяти. При аварийном завершении процесса принятые задачи будут потеряны. Задача, которая игнорирует свой контекст, может задержать Stop, поскольку Go не умеет принудительно завершать горутину. Если нужны восстановление после сбоя или повторные попытки, используйте надёжную персистентную очередь задач, например Asynq.

Полная реализация доступна в репозитории с примером.

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

• 9 сентября, 20:00. «Go‑профилирование: как найти и исправить „тормоза“ в продакшене». Записаться

• 22 сентября, 20:00. «Горутины и каналы: под капотом и нюансы в продакшене». Записаться

Полный список открытых уроков на конец августа и начало сентября собрали в дайджесте.