2.7Конкурентные паттерны

Соберём worker pool и pipeline с ограниченной нагрузкой, отменой и понятным владельцем закрытия каналов.

Из чего собираются паттерны

Worker pool

Задача: обработать M элементов не более чем N параллельными обработчиками, собрать результаты и ошибки, уметь остановиться по сигналу в любой момент. Самый частый запрос на лайвкодинге в Go. Спотыкаются почти все в трёх местах: кто закрывает канал задач, кто закрывает канал результатов и что происходит с отправкой результата, если получатель уже ушёл.

type Job struct {
    ID      int
    Payload string
}

type Result struct {
    JobID int
    Out   string
    Err   error
}

// Run обрабатывает jobs не более чем workers горутинами.
// Возвращает все результаты (включая ошибочные) и причину досрочного выхода.
func Run(ctx context.Context, workers int, jobs []Job) ([]Result, error) {
    jobCh := make(chan Job)
    // Буфер на все результаты: воркер никогда не заблокируется на отправке,
    // даже если коллектор отстаёт. Это снимает целый класс утечек.
    resCh := make(chan Result, len(jobs))

    var wg sync.WaitGroup
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := range jobCh {          // выход ровно при close(jobCh)
                out, err := handle(ctx, j)
                select {
                case resCh <- Result{JobID: j.ID, Out: out, Err: err}:
                case <-ctx.Done():          // страховка, если буфера всё же не хватило
                    return
                }
            }
        }(w)
    }

    // Продюсер единолично владеет jobCh и сам его закрывает.
    go func() {
        defer close(jobCh)                  // defer: закроется и на раннем return
        for _, j := range jobs {
            select {
            case jobCh <- j:
            case <-ctx.Done():
                return                      // раздачу прекратили, начатое доработается
            }
        }
    }()

    // Закрыть resCh вправе только тот, кто дождался всех писателей.
    go func() {
        wg.Wait()
        close(resCh)
    }()

    out := make([]Result, 0, len(jobs))
    for r := range resCh {                  // выход по close(resCh)
        out = append(out, r)
    }
    return out, ctx.Err()                   // nil, если отмены не было
}

Разбор по строкам

• for j := range jobCh вместо for { select { case j := <-jobCh } }. Короче и корректнее: range сам завершается при close, и не надо руками разбирать ok. Отдельная ветка на ctx.Done() в цикле воркера не нужна: отмену ловит продюсер, он просто перестаёт раздавать задачи, а «прервать посередине текущей задачи» это дело handle(ctx, j).
• wg.Add(1) до go, а не внутри. Внутри горутины Add может выполниться уже после Wait: гонка, при которой Wait вернётся раньше времени. В Go 1.25 весь этот блок ужимается до wg.Go(func(){ ... }), который делает Add(1), запуск и Done() одним вызовом.
• Буфер len(jobs) у resCh. Воркер не застрянет на отправке ни при каких условиях. Если задач может быть миллион, буфер берут небольшой (например, workers*2), и тогда ветка case <-ctx.Done() при отправке уже обязательна, а не страховочная.
• Отдельная горутина под wg.Wait(); close(resCh). Напишешь wg.Wait() прямо перед range resCh, и воркеры встанут на отправке в заполненный или небуферизованный канал, а main застрянет на Wait. Взаимная блокировка.
• Ошибки как поле Result, а не отдельный канал. Отдельный errCh тянет за собой второго читателя и вторую точку закрытия; результат с полем Err не тянет ничего.

Долгоживущий пул как компонент

// В реальном сервисе чаще так: пул живёт всё время работы
// приложения, задачи приходят снаружи, останавливается вместе с приложением.
type Pool struct {
    jobs chan Job
    wg   sync.WaitGroup
    once sync.Once
}

func NewPool(ctx context.Context, workers, queue int, h Handler) *Pool {
    p := &Pool{jobs: make(chan Job, queue)}
    for i := 0; i < workers; i++ {
        p.wg.Add(1)
        go func() {
            defer p.wg.Done()
            for j := range p.jobs {
                func() {
                    defer func() {
                        if r := recover(); r != nil {
                            log.Error("паника в воркере", "job", j.ID, "panic", r,
                                "stack", string(debug.Stack()))
                        }
                    }()
                    h(ctx, j)
                }()
            }
        }()
    }
    return p
}

// Submit не блокируется навсегда: либо очередь примет, либо отмена, либо отказ.
func (p *Pool) Submit(ctx context.Context, j Job) error {
    select {
    case p.jobs <- j:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

// SubmitOrDrop с backpressure: лучше отказать, чем копить очередь.
func (p *Pool) SubmitOrDrop(j Job) error {
    select {
    case p.jobs <- j:
        return nil
    default:
        return ErrQueueFull        // отдать 429 клиенту, инкрементировать метрику
    }
}

// Stop идемпотентен и дожидается доработки очереди.
func (p *Pool) Stop() {
    p.once.Do(func() { close(p.jobs) })   // once: защита от повторного close
    p.wg.Wait()
}

Pipeline: стадии через каналы

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

// Стадия 0: генератор. Принимает данные, отдаёт канал.
func gen(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)                 // владелец всегда закрывает свой канал
        for _, n := range nums {
            select {
            case out <- n:               // отправка тоже через select
            case <-ctx.Done():
                return
            }
        }
    }()
    return out                           // <-chan: наружу только чтение
}

// Стадия 1: преобразование.
func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {              // завершится, когда in закроют
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Стадия 2: фильтр. Сигнатура та же, стадии можно тасовать как кубики.
func odd(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if n%2 == 0 {
                continue
            }
            select {
            case out <- n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()                       // остановит весь конвейер при любом выходе

    for v := range odd(ctx, square(ctx, gen(ctx, 1, 2, 3, 4, 5))) {
        fmt.Println(v)
        if v > 8 {
            cancel()                     // досрочная остановка, всё свернётся само
            break
        }
    }
}

// вывод (проверено запуском), строк две, а не три:
// 1
// 9
// Конвейер выдал бы 1, 9, 25 (нечётные квадраты), но на 9 срабатывает v > 8:
// потребитель зовёт cancel() и выходит из range, а 25 до него уже не доезжает.

Разбор

• Единая сигнатура func(ctx, <-chan T) <-chan T и есть «конвейер»: стадии складываются друг с другом, их можно переставлять и вставлять новые, не трогая соседей.
• defer close(out) первой строкой горутины. Канал закроется на любом пути выхода, включая ранний return по отмене. Без него отмена посередине оставит следующую стадию в вечном range.
• select на отправке нужен не для красоты. Если следующая стадия умерла (отмена, паника, ранний break у потребителя), обычная отправка out <- v заблокируется навсегда, и вся цепочка утечёт. Именно эту строку чаще всего забывают.
• break у потребителя обязан сопровождаться cancel(). Иначе выход из range прекращает чтение, а стадии остаются на отправке. Идиома defer cancel() в вызывающем коде закрывает вопрос сама.
• Буфер на стадии: тюнинг, а не архитектура. Буфер сглаживает разницу в скорости соседних стадий, но не лечит несбалансированный конвейер: если стадия медленнее входа, буфер лишь отложит момент, когда всё упрётся в неё.

Fan-out / fan-in и своя реализация merge

Fan-out: несколько горутин читают из одного канала, стадия распараллеливается. Fan-in: несколько каналов сливаются в один (merge). Написать merge руками просят особенно охотно: в двадцати строках помещаются и владение каналом, и WaitGroup, и отмена.

// merge, каноническая реализация fan-in. Тридцать секунд на доске.
func merge[T any](ctx context.Context, chans ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup

    output := func(c <-chan T) {
        defer wg.Done()
        for v := range c {               // завершится при close(c)
            select {
            case out <- v:
            case <-ctx.Done():           // получатель ушёл, не виснем
                return
            }
        }
    }

    wg.Add(len(chans))                   // Add до запуска горутин
    for _, c := range chans {
        go output(c)
    }

    // Закрывает тот и только тот, кто дождался всех писателей.
    go func() {
        wg.Wait()
        close(out)
    }()

    return out                           // возврат немедленный, до окончания работы
}

// Fan-out поверх той же стадии: N копий читают один вход, их выходы сливаются.
func fanOut[T, R any](ctx context.Context, in <-chan T, n int,
    stage func(context.Context, <-chan T) <-chan R) <-chan R {

    outs := make([]<-chan R, 0, n)
    for i := 0; i < n; i++ {
        outs = append(outs, stage(ctx, in))   // все n читают из одного in
    }
    return merge(ctx, outs...)
}

// Применение: медленная стадия распараллелена на 8, порядок при этом теряется.
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    src := gen(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    for v := range fanOut(ctx, src, 8, slowSquare) {
        fmt.Println(v)                   // порядок произвольный
    }
}

// Три прогона подряд (значения всегда те же десять, порядок каждый раз новый):
//   36 64 1 49 9 4 25 16 100 81
//   64 49 1 4 16 36 25 9 100 81
//   64 49 36 16 9 4 25 1 100 81
// Полагаться на порядок нельзя: если он нужен, нумеруют элементы и пересобирают
// результат на выходе.

Разбор

• Почему wg.Add(len(chans)) одним вызовом. Иначе первая горутина успеет завершиться и уронить счётчик в ноль раньше, чем добавят остальные: Wait вернётся преждевременно и закроет out под живыми писателями, а это паника send on closed channel.
• Fan-out и есть «n читателей одного канала». Канал уже потокобезопасен и раздаёт элементы строго по одному: никакой дополнительной синхронизации не нужно. Это самая изящная часть модели каналов, и её стоит проговорить вслух.
• Балансировка получается сама. Кто освободился, тот и забрал следующий элемент: естественный work stealing без единой строчки кода. Медленный воркер просто получит меньше задач.
• Дженерики здесь по делу. До 1.18 merge писали на interface{} либо копипастили под каждый тип; сейчас одна функция закрывает всё.

Семафор: ограничить число одновременных запросов

Отличие от worker pool принципиальное, и его любят проверять: пул ограничивает число горутин, а семафор ограничивает число одновременно выполняемых операций при каком угодно числе горутин. Когда запросы приходят снаружи (каждый HTTP-хендлер живёт в своей горутине) и надо всего лишь не бомбить внешний API больше чем в 10 потоков, пул не годится: горутины созданы уже не вами. Годится семафор.

// Семафор на буферизованном канале: ёмкость буфера и есть лимит.
type Semaphore chan struct{}

func NewSemaphore(n int) Semaphore { return make(Semaphore, n) }

// Acquire ждёт слот или отмену, навсегда не блокируется.
func (s Semaphore) Acquire(ctx context.Context) error {
    select {
    case s <- struct{}{}:              // занял слот
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

// TryAcquire не ждёт: либо сразу, либо отказ (backpressure).
func (s Semaphore) TryAcquire() bool {
    select {
    case s <- struct{}{}:
        return true
    default:
        return false
    }
}

func (s Semaphore) Release() { <-s }   // освободил слот

// Применение: 10 одновременных вызовов внешнего API на весь процесс.
var apiSem = NewSemaphore(10)

func CallAPI(ctx context.Context, req Request) (*Response, error) {
    if err := apiSem.Acquire(ctx); err != nil {
        return nil, fmt.Errorf("очередь к API переполнена: %w", err)
    }
    defer apiSem.Release()             // только через defer, иначе слот утечёт навсегда
    return doHTTP(ctx, req)
}

Разбор

• Почему chan struct{}, а не chan bool. struct{} занимает ноль байт; для буфера на 1000 слотов это разница между нулём и килобайтом. Плюс сигнальная семантика читается сразу.
• Отправка = захват, приём = освобождение. Ровно наоборот интуиции, поэтому направление стоит проговорить: слоты «складываются» в буфер, полный буфер = все слоты заняты.
• Симметрия Acquire/Release держится на defer. Один return без Release, и лимит навсегда уменьшился на единицу. Через несколько таких утечек сервис встаёт с виду беспричинно.
• Fairness. Очередь ожидающих на канале работает по FIFO, поэтому голодания нет: кто раньше встал, тот раньше получит слот.

// Взвешенный вариант из стандартного расширения: разные операции стоят по-разному.
import "golang.org/x/sync/semaphore"

// Ёмкость в «единицах веса», например мегабайтах памяти.
var sem = semaphore.NewWeighted(1024)

func Process(ctx context.Context, f *File) error {
    w := int64(f.SizeMB)
    if w > 1024 {
        return ErrTooLarge             // Acquire с весом больше ёмкости только ждал бы отмены ctx
    }
    if err := sem.Acquire(ctx, w); err != nil {
        return err
    }
    defer sem.Release(w)
    return decode(ctx, f)
}
// Отличия от канального: веса, честная FIFO-очередь ожидающих, TryAcquire в комплекте.
// Цена: мьютекс и список ожидающих вместо одного канала. Чуть дороже, зато гибче.

Rate limiter: тикер и токен-бакет

Два способа. На тикере: жёсткий равномерный интервал, всплески невозможны в принципе. На токенах: средняя частота держится, но накопленные токены разрешают короткий всплеск.

Токен-бакет («ведро с токенами») устроен ровно так, как называется: есть ведро ёмкостью N, в него с постоянной скоростью капают токены, а каждый запрос перед отправкой забирает один. Токенов нет, ждёшь, пока накапает. Если запросов долго не было, ведро наполнилось, и следующие N запросов уйдут разом; это и есть всплеск (burst). Ёмкость ведра задаёт максимальный размер всплеска, скорость капания задаёт среднюю частоту. Второй вариант почти всегда ближе к реальности: внешние API обычно разрешают короткий burst, лишь бы среднее держалось в пределах лимита.

// Вариант 1: на тикере. Ровно N событий в секунду, без всплесков.
type TickLimiter struct {
    t    *time.Ticker
    done chan struct{}
}

func NewTickLimiter(rps int) *TickLimiter {
    return &TickLimiter{
        t:    time.NewTicker(time.Second / time.Duration(rps)),
        done: make(chan struct{}),
    }
}

func (l *TickLimiter) Wait(ctx context.Context) error {
    select {
    case <-l.t.C:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    case <-l.done:
        return ErrClosed
    }
}

func (l *TickLimiter) Close() { l.t.Stop(); close(l.done) }

// Вариант 2: токен-бакет своими руками. Пополняется фоном, всплеск до burst.
type Bucket struct {
    tokens chan struct{}
}

func NewBucket(ctx context.Context, rps, burst int) *Bucket {
    b := &Bucket{tokens: make(chan struct{}, burst)}
    for i := 0; i < burst; i++ {
        b.tokens <- struct{}{}                 // бакет стартует полным
    }
    go func() {
        t := time.NewTicker(time.Second / time.Duration(rps))
        defer t.Stop()
        for {
            select {
            case <-t.C:
                select {
                case b.tokens <- struct{}{}:   // положили токен
                default:                       // бакет полон, токен «сгорает»
                }
            case <-ctx.Done():
                return                         // без этого горутина утечёт
            }
        }
    }()
    return b
}

func (b *Bucket) Wait(ctx context.Context) error {
    select {
    case <-b.tokens:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (b *Bucket) Allow() bool {                // неблокирующий вариант
    select {
    case <-b.tokens:
        return true
    default:
        return false
    }
}

lim := rate.NewLimiter(rate.Limit(100), 20)   // 100 rps, всплеск до 20

if err := lim.Wait(ctx); err != nil { ... }   // подождать своей очереди
if !lim.Allow() { return errTooMany }         // отказать сразу
r := lim.Reserve()                            // узнать, сколько ждать, и решить
if !r.OK() { return errTooMany }
time.Sleep(r.Delay())

// Почему лучше самописного: нет фоновой горутины вообще. Токены не хранятся --
// они ВЫЧИСЛЯЮТСЯ из времени последнего обращения (lazy refill). Нечего
// останавливать, нечему утекать, точность не зависит от разрешения тикера.

Generator, or-done, tee

Три микропаттерна из «Concurrency in Go». Их спрашивают на понимание, а не на объём кода. Все три о том, как обернуть канал, чтобы им было безопасно пользоваться. generator («генератор») превращает произвольный источник данных в канал. or-done (дословно «или готово») оборачивает чужой канал и добавляет ему отменяемость, чтобы читатель мог уйти, даже если владелец канала об отмене не знает. tee (по названию T-образного сантехнического тройника) раздваивает поток: один входной канал, два выходных, в оба уходит каждое значение.

// Generator превращает что угодно в канал. Ленивый, отменяемый, бесконечный.
func repeat[T any](ctx context.Context, vals ...T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {                                  // бесконечный источник
            for _, v := range vals {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

// take ограничивает бесконечный генератор
func take[T any](ctx context.Context, in <-chan T, n int) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for i := 0; i < n; i++ {
            select {
            case v, ok := <-in:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Or-done: оборачивает чужой канал, за поведение которого не отвечаешь,
// и делает его отменяемым. Вместо select хватает простого range.
func orDone[T any](ctx context.Context, in <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                return
            case v, ok := <-in:
                if !ok {
                    return                     // источник закрылся сам
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}
// Смысл: вместо громоздкого select в каждом потребителе пишем
//   for v := range orDone(ctx, foreignCh) { ... }
// и гарантированно не залипаем, если источник перестал закрывать канал.

// Tee раздваивает поток: один вход, два независимых выхода.
func tee[T any](ctx context.Context, in <-chan T) (<-chan T, <-chan T) {
    out1, out2 := make(chan T), make(chan T)
    go func() {
        defer close(out1)
        defer close(out2)
        for v := range orDone(ctx, in) {
            // Локальные копии: обнуляем отправленный канал, чтобы select
            // не выбрал его дважды. Пока оба не nil, крутимся.
            o1, o2 := out1, out2
            for i := 0; i < 2; i++ {
                select {
                case <-ctx.Done():
                    return
                case o1 <- v:
                    o1 = nil                   // отправка в nil блокируется навсегда,
                case o2 <- v:                  // значит этот case выключен
                    o2 = nil
                }
            }
        }
    }()
    return out1, out2
}

Отмена по первой ошибке против сбора всех ошибок

Два взаимоисключающих режима, и по любой параллельной задаче первым делом спрашивают, какой из них нужен. «Первая ошибка отменяет всё» берут, когда результат бесполезен без каждой части: собрать страницу из трёх источников. «Собрать все ошибки» берут, когда части независимы и надо знать полную картину: валидация формы, рассылка уведомлений, батч-импорт.

// Режим 1: errgroup, первая ошибка отменяет остальных.
import "golang.org/x/sync/errgroup"

func loadPage(ctx context.Context, id int) (*Page, error) {
    g, ctx := errgroup.WithContext(ctx)      // ctx переприсваиваем намеренно
    var p Page

    g.Go(func() error {
        var err error
        p.User, err = fetchUser(ctx, id)     // получает уже производный ctx
        return err
    })
    g.Go(func() error {
        var err error
        p.Feed, err = fetchFeed(ctx, id)
        return err
    })
    g.Go(func() error {
        var err error
        p.Ads, err = fetchAds(ctx, id)
        return err
    })

    if err := g.Wait(); err != nil {         // вернёт первую по времени ошибку
        return nil, err
    }
    return &p, nil
}

// Ограничение параллелизма встроено:
g.SetLimit(8)                                // не больше 8 одновременно
// TryGo не блокирует: false, если лимит исчерпан
if !g.TryGo(fn) { /* очередь занята */ }

// Режим 2: собрать все ошибки. errgroup здесь не подходит.
func notifyAll(ctx context.Context, users []User) error {
    var (
        mu   sync.Mutex
        errs []error
        wg   sync.WaitGroup
    )
    sem := make(chan struct{}, 16)           // параллелизм ограничиваем сами

    for _, u := range users {
        wg.Add(1)
        go func(u User) {
            defer wg.Done()
            sem <- struct{}{}
            defer func() { <-sem }()

            if err := notify(ctx, u); err != nil {
                mu.Lock()
                errs = append(errs, fmt.Errorf("user %d: %w", u.ID, err))
                mu.Unlock()
            }
        }(u)
    }
    wg.Wait()
    return errors.Join(errs...)              // Go 1.20: nil, если срез пуст
}

Периодические задачи и корректная остановка

// Канонический периодический воркер.
func (s *Service) runPeriodic(ctx context.Context, every time.Duration) {
    t := time.NewTicker(every)
    defer t.Stop()                       // до Go 1.23 без этого тикер жил вечно

    s.tick(ctx)                          // часто нужен сразу первый прогон,
                                         // Ticker выдаёт первый тик только через every
    for {
        select {
        case <-t.C:
            s.tick(ctx)                  // tick должен уважать ctx и иметь свой таймаут
        case <-ctx.Done():
            return                       // выход по общей отмене
        }
    }
}

func (s *Service) tick(ctx context.Context) {
    // Свой дедлайн на итерацию: иначе одна залипшая итерация останавливает всё.
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()

    defer func() {                       // паника в фоновой задаче не должна ронять сервис
        if r := recover(); r != nil {
            log.Error("паника в периодической задаче", "panic", r,
                "stack", string(debug.Stack()))
        }
    }()

    if err := s.doWork(ctx); err != nil {
        log.Warn("итерация не удалась", "err", err)   // не return из runPeriodic:
    }                                                 // упавшая итерация не должна
}                                                     // убивать расписание

Graceful shutdown приложения целиком

Самый «взрослый» вопрос главы. Знание API тут вторично: проверяют, понимаешь ли ты, что сервис собран из компонентов с зависимостями и останавливать их надо в строго определённом порядке, снаружи внутрь. Сначала перестаём принимать новую работу, потом доделываем начатую, и только в самом конце закрываем то, что этой работе нужно. Перевернёшь порядок, получишь гарантированные ошибки «connection closed» в последних запросах.

func main() {
    // 1. Корневой контекст, отменяемый по сигналу. Второй SIGINT вернёт
    //    поведение по умолчанию, то есть немедленное убийство процесса.
    ctx, stop := signal.NotifyContext(context.Background(),
        syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    // Инициализация: обратный порядок закрытия задаётся здесь.
    db := mustOpenDB(ctx)
    cache := mustOpenRedis(ctx)
    pool := NewPool(ctx, 16, 1024, handleJob)
    consumer := NewKafkaConsumer(cfg, pool)
    srv := &http.Server{Addr: ":8080", Handler: newRouter(db, cache, pool)}

    var ready atomic.Bool
    ready.Store(true)
    http.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
        if !ready.Load() {
            w.WriteHeader(http.StatusServiceUnavailable)
            return
        }
        w.WriteHeader(http.StatusOK)
    })

    go func() {
        if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
            log.Error("сервер упал", "err", err)
            stop()                                  // роняем всё приложение штатно
        }
    }()
    go consumer.Run(ctx)

    <-ctx.Done()                                    // ждём сигнала
    stop()                                          // вернуть обработчик по умолчанию:
    log.Info("получен сигнал, останавливаемся")     // второй Ctrl+C убьёт сразу

    // 2. Общий бюджет на всю остановку. Он должен быть меньше
    //    terminationGracePeriodSeconds, иначе процесс добьют SIGKILL.
    shutCtx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
    defer cancel()

    // 3. Снимаем себя с балансировки и ждём, пока правила разъедутся.
    ready.Store(false)
    select {
    case <-time.After(7 * time.Second):
    case <-shutCtx.Done():
    }

    // 4. HTTP: новые соединения не принимаем, текущие дорабатывают.
    if err := srv.Shutdown(shutCtx); err != nil {
        log.Warn("не все запросы успели, рвём принудительно", "err", err)
        _ = srv.Close()
    }

    // 5. Консьюмер: перестаём брать новые сообщения, коммитим оффсеты.
    consumer.Stop(shutCtx)

    // 6. Фоновые воркеры: очередь дорабатывается, новые задачи не принимаются.
    done := make(chan struct{})
    go func() { pool.Stop(); close(done) }()
    select {
    case <-done:
    case <-shutCtx.Done():
        log.Warn("воркеры не успели за отведённое время")
    }

    // 7. Флашим телеметрию до закрытия соединений, но после остановки писателей.
    _ = tracerProvider.Shutdown(shutCtx)
    metrics.Flush(shutCtx)

    // 8. Ресурсы закрываем последними, в порядке, обратном открытию.
    _ = cache.Close()
    _ = db.Close()

    log.Info("остановлено штатно")
}

Разбор ключевых решений

• signal.NotifyContext вместо signal.Notify + канала. Сразу даёт отменяемый ctx, который можно раздать всем компонентам, и корректно снимает обработчик через stop().
• Вызов stop() сразу после <-ctx.Done(). Возвращает сигналам поведение по умолчанию: если оператор нервничает и жмёт Ctrl+C второй раз, процесс умрёт немедленно, а не проигнорирует сигнал.
• Отдельный shutCtx от context.Background(). Корневой ctx уже отменён: возьмёшь его, и все шаги остановки завершатся мгновенно с ошибкой, никакого graceful. Это ошибка номер один у тех, кто пишет shutdown впервые.
• Общий бюджет, а не таймаут на каждый шаг. Один shutCtx на всё гарантирует, что суммарное время не превысит договорённость с оркестратором.
• Порядок закрытия ресурсов обратен порядку открытия. БД открыли первой, закрываем последней: от неё зависят все остальные.
• pool.Stop() обёрнут в горутину с select. Сам Stop блокирующий и контекста не принимает; без обёртки зависший воркер сорвёт весь бюджет и дождётся SIGKILL.

jobs := make(chan Job)
res  := make(chan Result, len(input))
var wg sync.WaitGroup

for i := 0; i < n; i++ {
    wg.Go(func() {                       // Go 1.25; до неё: wg.Add(1) + go func + defer wg.Done()
        for j := range jobs {
            out, err := handle(ctx, j)
            res <- Result{j.ID, out, err}
        }
    })
}
go func() { defer close(jobs); for _, j := range input { jobs <- j } }()
go func() { wg.Wait(); close(res) }()

for r := range res { collect(r) }

// Если без закрытия извне никак, вот безопасный враппер.
type SafeChan[T any] struct {
    mu     sync.RWMutex
    closed bool
    ch     chan T
}

func (s *SafeChan[T]) Send(v T) error {
    s.mu.RLock()                         // читатели-отправители не мешают друг другу
    defer s.mu.RUnlock()
    if s.closed {
        return ErrClosed
    }
    s.ch <- v                            // под RLock: Close не сможет вклиниться
    return nil
}

func (s *SafeChan[T]) Close() {
    s.mu.Lock()                          // ждёт, пока все Send завершатся
    defer s.mu.Unlock()
    if !s.closed {
        s.closed = true
        close(s.ch)
    }
}
// Цена: RLock на каждую отправку и риск задержки Close, если канал полон,
// а читателя нет. Поэтому это запасной вариант, а не основной подход.

// Практичный вариант: ошибка едет по конвейеру вместе с данными.
type Item[T any] struct {
    Val T
    Err error
}

func stage[T, R any](ctx context.Context, in <-chan Item[T],
    f func(context.Context, T) (R, error)) <-chan Item[R] {

    out := make(chan Item[R])
    go func() {
        defer close(out)
        for it := range in {
            var res Item[R]
            if it.Err != nil {
                res = Item[R]{Err: it.Err}     // ошибку пропускаем дальше как есть
            } else {
                v, err := f(ctx, it.Val)
                res = Item[R]{Val: v, Err: err}
            }
            select {
            case out <- res:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}
// Плюс: один канал, одна точка закрытия, ошибки не теряются и видно, на каком
// элементе сломалось. Минус: каждая стадия обязана проверять it.Err первой строкой.

func merge2[T any](ctx context.Context, a, b <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for a != nil || b != nil {           // пока жив хоть один источник
            select {
            case v, ok := <-a:
                if !ok {
                    a = nil                  // выключаем ветку навсегда
                    continue
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            case v, ok := <-b:
                if !ok {
                    b = nil
                    continue
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}
// Почти все забывают обнулить закрытый канал.
// Закрытый канал всегда готов к приёму, select начнёт выбирать его
// в бесконечном цикле: busy-loop на 100% CPU вместо аккуратного слияния.

// Самый простой и самый недооценённый вариант: канал результатов не нужен вообще.
// Каждая горутина пишет в свою ячейку, гонки нет, синхронизация не нужна.
func mapParallel[T, R any](ctx context.Context, in []T, limit int,
    f func(context.Context, T) (R, error)) ([]R, error) {

    out := make([]R, len(in))
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(limit)

    for i, v := range in {
        g.Go(func() error {
            r, err := f(ctx, v)
            if err != nil {
                return fmt.Errorf("элемент %d: %w", i, err)
            }
            out[i] = r                       // своя ячейка, без синхронизации
            return nil
        })
    }
    if err := g.Wait(); err != nil {         // Wait как барьер: после него слайс виден целиком
        return nil, err
    }
    return out, nil
}
// Почему запись в разные элементы слайса безопасна: это разные адреса памяти.
// Слайс не перевыделяется (длина фиксирована), заголовок никто не меняет.
// А g.Wait() создаёт happens-before между записями и последующим чтением.

// Когда на входе бесконечный поток и слайс не построить.
type Seq[T any] struct {
    Idx int
    Val T
}

func reorder[T any](ctx context.Context, in <-chan Seq[T]) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        buf  := make(map[int]T)
        next := 0
        for it := range in {
            buf[it.Idx] = it.Val
            for {                            // отдаём всё, что стало непрерывным
                v, ok := buf[next]
                if !ok {
                    break
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
                delete(buf, next)
                next++
            }
        }
    }()
    return out
}
// Цена честного порядка: один застрявший элемент держит в памяти всё,
// что обогнало его. Обязательно ограничивать размер buf и падать/тормозить
// при переполнении, иначе медленный элемент превращается в OOM.

type Semaphore chan struct{}

func NewSemaphore(n int) Semaphore { return make(Semaphore, n) }

func (s Semaphore) Acquire(ctx context.Context) error {
    select {
    case s <- struct{}{}:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}
func (s Semaphore) TryAcquire() bool {
    select {
    case s <- struct{}{}:
        return true
    default:
        return false
    }
}
func (s Semaphore) Release() { <-s }
func (s Semaphore) InUse() int { return len(s) }   // готовая метрика насыщения

// Лимитер на клиента: то, что нужно в 90% случаев.
type PerKey struct {
    mu   sync.Mutex
    lim  map[string]*entry
    r    rate.Limit
    b    int
}
type entry struct {
    l    *rate.Limiter
    seen time.Time
}

func (p *PerKey) Allow(key string) bool {
    p.mu.Lock()
    e, ok := p.lim[key]
    if !ok {
        e = &entry{l: rate.NewLimiter(p.r, p.b)}
        p.lim[key] = e
    }
    e.seen = time.Now()
    p.mu.Unlock()
    return e.l.Allow()
}

// Без вытеснения карта растёт вечно: это утечка памяти, а не «мелочь».
func (p *PerKey) gc(ctx context.Context, ttl time.Duration) {
    t := time.NewTicker(ttl)
    defer t.Stop()
    for {
        select {
        case <-t.C:
            p.mu.Lock()
            for k, e := range p.lim {
                if time.Since(e.seen) > ttl {
                    delete(p.lim, k)
                }
            }
            p.mu.Unlock()
        case <-ctx.Done():
            return
        }
    }
}

// Было: в каждом потребителе шум, который легко забыть.
loop:
for {
    select {
    case v, ok := <-foreign:
        if !ok {
            break loop
        }
        use(v)
    case <-ctx.Done():
        break loop
    }
}

// Стало: защита один раз, потребители тривиальны.
for v := range orDone(ctx, foreign) {
    use(v)
}

type Group struct {
    cancel  func(error)
    wg      sync.WaitGroup
    sem     chan token      // появляется после SetLimit
    errOnce sync.Once       // «первая ошибка» фиксируется ровно один раз
    err     error
}

func (g *Group) Go(f func() error) {
    if g.sem != nil {
        g.sem <- token{}    // SetLimit сводится к семафору на канале
    }
    g.wg.Add(1)
    go func() {
        defer g.done()
        if err := f(); err != nil {
            g.errOnce.Do(func() {
                g.err = err
                if g.cancel != nil {
                    g.cancel(g.err)   // отменяем производный ctx, остальные сворачиваются
                }
            })
        }
    }()
}

g, _ := errgroup.WithContext(ctx)   // ПОТЕРЯЛИ производный ctx
g.Go(func() error { return fetch(ctx, a) })   // получает ИСХОДНЫЙ ctx
// Ошибка вернётся, но остальные горутины доработают до конца:
// отменять-то нечего, производный контекст выброшен.

g, ctx := errgroup.WithContext(ctx)  // ПРАВИЛЬНО: ctx теперь производный
g.Go(func() error { return fetch(ctx, a) })

// Ticker меряет период между тиками. Если нужен фиксированный интервал
// между окончанием и началом (работа длится непредсказуемо долго), берут Timer.
func runFixedGap(ctx context.Context, gap time.Duration, f func(context.Context)) {
    for {
        f(ctx)
        select {
        case <-time.After(gap):     // с Go 1.23 таймер соберётся GC, утечки нет
        case <-ctx.Done():
            return
        }
    }
}

// Ticker отбрасывает пропущенные тики, а не копит их.
// Это защита от лавины: если итерация длилась 10 периодов, то после неё
// придёт один тик, а не 10 подряд. Но и фактическая частота упадёт:
// иногда это ровно то, что нужно, иногда это скрытая деградация.

// Drop-oldest: в очереди всегда самое свежее.
func (q *Queue) PushLatest(v Item) {
    for {
        select {
        case q.ch <- v:
            return
        default:
            select {
            case <-q.ch:            // выбросили самый старый
                q.dropped.Add(1)    // метрика обязательна: без неё потери невидимы
            default:                // кто-то успел забрать, пробуем снова
            }
        }
    }
}

Что стоит запомнить

• Worker pool ограничивает количество параллельных обработчиков.
• Закрывать output можно после завершения всех sender.
• Semaphore ограничивает concurrency, limiter - частоту.

Частые ошибки

• Consumer cancellation должна доходить до заблокированного producer.
• Не оставляй бесконечное число созданных goroutine ожидать лимит.

jobs := make(chan int); results := make(chan int)
var wg sync.WaitGroup
for i:=0;i<2;i++ { wg.Add(1); go func(){ defer wg.Done(); for n := range jobs { results <- n*n } }() }
go func(){ wg.Wait(); close(results) }()
go func(){ for _, n := range []int{2,3} { jobs <- n }; close(jobs) }()
for r := range results { fmt.Println(r) }

for _, job := range jobs { go func() { results <- work(job) }() }

jobsCh := make(chan Job, 32)
results := make(chan Result, 32)
var wg sync.WaitGroup
for i := 0; i < workerCount; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for job := range jobsCh {
            result := work(job)
            select { case results <- result: case <-ctx.Done(): return }
        }
    }()
}
go func() { wg.Wait(); close(results) }()
producer:
for _, job := range input {
    select { case jobsCh <- job: case <-ctx.Done(): break producer }
}
close(jobsCh)