Простой

9 мин

28

Часто про Go говорят: «это язык, где конкурентность почти бесплатная».

И знаете что? Это правда. Почти.

Но «почти» — это самое опасное во всей истории, так как либо ты управляешь системой, либо она управляет тобой руками runtime'а.

В трёх статьях я разберу путь, через который проходит почти каждый Go-разработчик от наивного «я добавил go — получил параллельность», до взрослого «я проектирую concurrency-систему с понятными границами».

Погнали.

Иллюзия первая: «Горутины дешёвые — значит можно сколько угодно»

Новички в Go рассуждают примерно так:

Горутина весит 2 КБ стека. Поток — 1 МБ. Значит, я могу создать 500 000 горутин вместо 2000 потоков. Отлично, пишу go на каждый чих!

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

Ниже приведен совершенно классический код:

func handleRequest(req Request) {
    go processAsync(req)  // Эмитируем асинхронность
    respondOK()
}

Кажется, что код выглядит безобидно, однако, если присмотреться, здесь нет:

• лимита на количество одновременно работающих горутин

• очереди

• контроля завершения

• backpressure'а

Получается, что тут вы не управляете системой, а надеетесь, что runtime справится и он честно пытается.

В метриках это выглядит примерно так:

Число горутин:      1000 → 10000 → 50000
Время ответа:       50ms → 200ms → 800ms
CPU:                30%  → 70%   → 95% (полезной работы — всё меньше)
Ошибки:             0    → 0     → 0

То есть формально наш сервис жив, но мёртв для пользователя.

Заглянем под капот

Планировщик Go (GMP-модель: Goroutines, Machine, Processor) начинает страдать:

• G (горутины) — их слишком много, очередь runqueue растягивается

• M (основные потоки ОС) — пытаются всё вывезти

• P (процессоры логические) — мечутся между горутинами чаще, чем выполняют полезную работу

Если говорить простыми словами, то планировщик тратит больше времени на переключение между задачами, чем на выполнение самих задач, это как если бы вы меняли инструменты каждые 10 секунд вместо того, чтобы работать. Однако, здесь может быть использован паттерн worker pool + ограниченная очередь, варианты его реализации мы разберем в части 3. Код одной из реализаций приведу ниже:

type WorkerPool struct {
    tasks chan func()
}

func (p *WorkerPool) worker() {
    for task := range p.tasks {
        task()
    }
}

func NewWorkerPool(workers int, queueSize int) *WorkerPool {
    pool := &WorkerPool{
        tasks: make(chan func(), queueSize),
    }
    for i := 0; i < workers; i++ {
        go pool.worker()
    }
    return pool
}

func (p *WorkerPool) Submit(task func()) error {
    select {
    case p.tasks <- task:
        return nil
    default:
        return ErrQueueFull // backpressure — важнейшая вещь
    }
}

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

У нас есть канал tasks — это очередь задач. Туда мы будем складывать функции, которые нужно выполнить, далее создаём пул:

func (p *WorkerPool) worker() {
    for task := range p.tasks {
        task()
    }
}

func NewWorkerPool(workers int, queueSize int) *WorkerPool {
  pool := &WorkerPool{
    tasks: make(chan func(), queueSize),
  }
  for i := 0; i < workers; i++ {
    go pool.worker()
  }
  return pool
}

Что здесь происходит:

• queueSize — это максимальный размер очереди

• workers — сколько задач может выполняться одновременно

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

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

func (p *WorkerPool) Submit(task func()) error {
  select {
    case p.tasks <- task:
    return nil
    default:
    return ErrQueueFull
  }
}

Вот здесь происходит магия - мы пытаемся положить задачу в канал и, если в очереди есть место — задача принимается, однако, если очередь заполнена — сразу получаем ErrQueueFull и это принципиально важно.

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

Инцидент первый: «подождём секундочку »

Код ниже, ломает 50% наивных реализаций:

func main() {
    go doWork()
    time.Sleep(1 * time.Second)  // "Ну, за секунду точно успеет"
}

Ключевая проблема time.Sleep — это не ожидание выполнения, а случайная пауза, которая иногда совпадает с реальностью.

Почему под нагрузкой такой код даст сбой мгновенно? Это происходит из-за ряда причин:

• doWork() начинает тормозить

• 1 секунды перестаёт хватать

• горутины накапливаются

• программа завершается до завершения работы

Формально код «рабочий». Но в реальности — это просто отложенный баг.

Правильное ожидание

func main() {
    done := make(chan struct{})
    go func() {
        defer close(done)
        doWork()
    }()
    <-done  // Ждём реального завершения
}

done := make(chan struct{})

Это обычный канал, который используется как сигнал: «работа закончилась»

Тип struct{} выбран не случайно — он ничего не занимает в памяти, а нам не нужно передавать данные, нам нужен сам факт.

Еще одна важная деталь — defer close(done) как только doWork() закончится (не важно — успешно или с ошибкой) канал будет закрыт и это наш сигнал наружу об окончании.

Как происходит ожидание

<-done

Здесь main-поток просто блокируется и ждёт и разблокируется он только в одном случае — когда канал закроют и это реальное ожидание, а не «подождём секунду и надеемся».

Мы больше не гадаем:

• хватит ли времени

• успеет ли задача

• не зависла ли она

Чем это лучше time.Sleep

Sleep — это всегда предположение: «я думаю, этого времени хватит», а канал — это гарантия: «я точно знаю, что работа завершилась»

Можно также реализовать с контекстом (если работа может занять слишком много времени):

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    done := make(chan struct{})
    go func() {
        defer close(done)
        doWork()
    }()
    
    select {
    case <-done:
        fmt.Println("успели")
    case <-ctx.Done():
        fmt.Println("не успели, но не утекли")
    }
}

И здесь мы задаем правило: готов ждать максимум 5 секунд, тут очень важный момент без него система думает, что может ждать бесконечно

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

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

Инцидент второй: нагрузка выросла в 2 раза — латенси улетела

Симптомы, знакомые многим, сервис наш работает корректно, ошибок нет, падений сервиса тоже нет, однако у нас растет время ответа сервиса 50ms → 800–1000ms CPU: 90–100%

Например такая схема, убивающая production:

func handleFanOut(req Request) {
    results := make(chan Result, len(req.SubTasks))
    
    for _, task := range req.SubTasks {
        go func(t Task) {          // Нет лимита! Каждый запрос плодит N горутин
            res := callExternalService(ctx,t)
            results <- res
        }(task)
    }
    
    for i := 0; i < len(req.SubTasks); i++ {
        <-results
    }
}

Переводя это в цифры:

• 1 запрос → 10 горутин

• 100 запросов → 1000 горутин

• 1000 запросов → 10 000 горутин (работает… пока работает)

• 2000 запросов → 20 000 горутин — всё падает

Причём падает не в момент, когда вы ожидаете, а когда планировщик говорит: «Ребята, я больше не могу.»

Диагностика и как это увидеть

Одним из вариантов запуск в консоли:

# Посмотреть число горутин
curl http://localhost:6060/debug/pprof/goroutine?debug=1

# Или через go tool
go tool pprof http://localhost:6060/debug/pprof/goroutine

# След планировщика — мастхэв
GODEBUG=schedtrace=1000 ./your-service

Вывод schedtrace:

SCHED 1005ms: gomaxprocs=8 idleprocs=0 threads=11 spinningthreads=1 idlethreads=0 runqueue=152 [3 2 1 0 0 0 0 0]

Ключевое на что нужно обратить внимание на runqueue=152 — это значит, что 152 горутины ждут выполнения, то есть наш планировщик захлёбывается.

Исправление

func handleFanOutControlled(ctx context.Context, req Request) error {
    sem := semaphore.NewWeighted(10) // максимум 10 одновременных вызовов
    var wg sync.WaitGroup
    
    for _, task := range req.SubTasks {
        wg.Add(1)
        go func(t Task) {
            defer wg.Done()
            if err := sem.Acquire(ctx, 1); err != nil {
                return
            }
            defer sem.Release(1)
            
            callExternalService(ctx,t)
        }(task)
    }
    
    done := make(chan struct{})
    go func() {
        wg.Wait()
        close(done)
    }()
    
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Мы начинаем контролировать параллелизм, а не количество запросов. Наша функция берет один запрос, разбивает его на подзадачи (SubTasks) и обрабатывает их параллельно, но тут есть важная оговорка: не более 10 задач и плюс есть нормальное завершение и таймаут через context

sem := semaphore.NewWeighted(10)

В любой момент времени можно выполнить не более 10 задач остальные будут ждать.

Внутри горутины - это будет работать так:

if err := sem.Acquire(ctx, 1); err != nil {
    return
}
defer sem.Release(1)

• Acquire — «можно ли мне начать работу?»

• если лимит достигнут → ждём

• если context отменён → выходим

После завершенияRelease освобождает слот и следующая задача может начать выполняться. Мы не ограничиваем создание goroutines (они всё равно создаются), но ограничиваем реальный параллелизм выполнения. Это дешевле, чем бесконтрольный fan-out, но всё ещё может давать overhead при очень большом числе задач.

Ожидание задач проходит через классический механизм var wg sync.WaitGroup

wg.Add(1)
...
defer wg.Done()

Стоит подсветить важный нюанс: return ctx.Err() не останавливает автоматически горутины, они могут продолжать работать. Чтобы горутины реально останавливались, контекст должен прокидываться внутрь всех долгих операций (например: callExternalService(ctx, t)), иначе они продолжат выполняться даже после отмены.

Почему этот код — «правильный»

Потому что здесь есть всё, чего обычно не хватает:

1. Ограничение

Не больше 10 задач одновременно → система не захлёбывается

2. Ожидание

Мы реально знаем, когда всё закончилось

3. Таймаут

Мы не зависаем бесконечно

Инцидент третий: скрытая утечка горутин

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

func startWorker(jobs <-chan Job) {
    go func() {
        for job := range jobs {
            process(job)
        }
        // если канал не закроется — сюда никогда не попадём
    }()
}

На первый взгляд, выглядит безопасно, однако тут есть проблема: если канал jobs никто не закроет, горутина будет висеть вечно и каждый рестарт или вызов startWorker добавит новую.

Через неделю у нас 20k горутин, стек по 2-8 КБ каждая — уже ~100-150 МБ. Память растёт медленно и не бросается в глаза, потому что сами данные маленькие, но суммарно тысячи горутин начинают съедать десятки мегабайт, а вот планировщик Go начинает задыхаться. В pprof/goroutine мы увидим тысячи chan receive.

Правильное завершение

Правильный вариант должен начинаться с простой мысли: у горутины должен быть жизненный цикл. Исправляется это довольно просто:

type Worker struct {
    jobs   chan int
    done   chan struct{}
    wg     sync.WaitGroup
}

func (w *Worker) Stop() {
    close(w.jobs)   // <- ключевой момент
    <-w.done        // ждём, пока worker закончит
}

func (w *Worker) Start() {
    w.wg.Add(1)
    go func() {
        defer w.wg.Done()
        defer close(w.done)
        
        for job := range w.jobs { // закроется, когда канал закроют
            process(job)
        }
    }()
}

Тут мы указываем явно два сигнала:

type Worker struct {
    jobs   chan int
    done   chan struct{}
    wg     sync.WaitGroup
}

• jobs — откуда приходят задачи

• done — сигнал, что воркер полностью завершился

for job := range w.jobs — это нормальный паттерн, но он работает корректно только, если канал когда-нибудь закроют. Как только jobs закрывается — цикл сам завершится после этого: вызывается wg.Done() закрывается done → внешний мир узнаёт: «воркер реально закончился»

Ну и конечно: close(w.jobs) — мы явно говорим: «новых задач больше не будет», воркер дочитывает всё, что уже было в канале и выходит из цикла, закрывает done мы дожидаемся этого через <-w.done

Ну и конечно, перед тем как написать go func() — нужн задать себе два вопроса: Кто и когда закроет её канал? и Что произойдёт, если этого не случится?

Если не знаем ответа — закладываем явный context с таймаутом или канал stop с select.

Инцидент четвёртый: внешний сервис начал тормозить, и мы упали вместе с ним

Код из разряда «работает годами, а потом бац»:

func fetchData(url string) (*Response, error) {
    resp, err := http.Get(url)  // А что, если 30 секунд?
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    return parse(resp)
}

Представим вполне обычную ситуацию: внешний сервис, к которому мы обращаемся, внезапно начинает подвисать, то есть вместо ответа 50мс получаем - 20-40 секунд. А следом за ним и у нас начинает деградировать сервис: наши запрос не падают, они просто висят.

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

Самое неприятное — это не выглядит как авария. Нет красных алертов «всё умерло». Есть тихая деградация, которая разъедает систему.

И тут вступает в действие Железное правило

Любой вызов внешнего сервиса должен иметь таймаут. Даже если он «локальный». Даже если «99.99% времени отвечает за 5 мс».

func fetchDataSafe(url string) (*Response, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
    
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return nil, err
    }
    
    client := &http.Client{
        Timeout: 2 * time.Second,  // второй круг защиты
    }
    
    resp, err := client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    
    return parse(resp)
}

Два таймаута — не паранойя - это production. И если context ограничивает время жизни запроса, то http.Client.Timeout страхует на уровне клиента. Если один слой не сработал — сработает второй.

Как увидеть проблемы до того, как они увидят вас

1. Обязательные метрики

import "github.com/prometheus/client_golang/prometheus"

var (
    goroutines = prometheus.NewGaugeFunc(
        prometheus.GaugeOpts{Name: "go_goroutines_current"},
        func() float64 { return float64(runtime.NumGoroutine()) },
    )
    
    // И это обязательно
    scheduleLatency = prometheus.NewHistogram(...)
)

runtime.NumGoroutine() — это вообще одна из самых недооценённых метрик. В нормальной системе она ведёт себя примерно так: под нагрузкой выросла потом, вернулась обратно, но если она растёт и не падает, то это почти всегда сигнал, что что-то пошло не так.

scheduleLatency — ещё более продвинутая штука, если она начинает расти, то планировщик уже не справляется.

2. Профилирование в тестах

func TestNoGoroutineLeak(t *testing.T) {
    before := runtime.NumGoroutine()
    
    // ваш код
    
    time.Sleep(time.Second) // даём завершиться
    after := runtime.NumGoroutine()
    
    if after > before {
        t.Errorf("утечка: было %d, стало %d", before, after)
    }
}

Да, это не идеальный тест, это база и он может давать шум, но как ранний сигнал — работает отлично.

Идея простая: замерили количество горутин «до», затем прогнали сценарий и дали системе чуть времени всё закрыть и конечно, посмотрели, что осталось

3. pprof в каждом сервисе

Тут можно сказать, что пока всё хорошо — он не нужен. Когда становится плохо — без него почти невозможно понять, что происходит.

import _ "net/http/pprof"

go func() {
    log.Println(http.ListenAndServe("localhost:6060", nil))
}()

На практике открываем /debug/pprof/goroutine и видим, где реально зависли горутины, так нам за 2 минуты понятна причина, вместо того чтобы часами гадать.

Помните: горутина — не бесплатная абстракция, это задача, которая конкурирует за ограниченный ресурс время планировщика. И в следующий раз, когда увидите код без контекста, лимитов и таймаутов — знайте: это не баг. Это будущий инцидент, которого можно избежать уже сегодня.