Паттерны конкурентности в Go

На live coding в Go чаще всего просят: обработать N задач с ограничением параллелизма, собрать результаты в исходном порядке и корректно остановиться при ошибке или таймауте. Ниже набор кирпичиков.

Worker pool

Фиксированное число воркеров читают из общего канала задач:

func pool(ctx context.Context, jobs []int, workers int) []int {
    res := make([]int, len(jobs))
    idx := make(chan int)
    var wg sync.WaitGroup
    for range workers {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := range idx {
                res[i] = jobs[i] * jobs[i] // каждый пишет в свою ячейку: гонки нет
            }
        }()
    }
loop:
    for i := range jobs {
        select {
        case idx <- i:
        case <-ctx.Done():
            break loop
        }
    }
    close(idx)
    wg.Wait()
    return res
}

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

Pipeline

Цепочка стадий, каждая читает из входного канала и пишет в выходной, а по окончании закрывает свой выход:

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums { out <- n }
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in { out <- n * n }
    }()
    return out
}

Если потребитель может уйти раньше, каждой стадии нужен ctx или канал done, иначе горутины утекут на заблокированной отправке.

Fan-out и fan-in

Fan-out: несколько горутин читают из одного канала. Fan-in (merge): несколько каналов сливаются в один.

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, c := range cs {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for v := range c { out <- v }
        }()
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

Ключевая деталь: out закрывает отдельная горутина после того, как все писатели закончили.

errgroup

golang.org/x/sync/errgroup это WaitGroup с ошибками и отменой:

g, ctx := errgroup.WithContext(ctx)
g.SetLimit(8) // не больше 8 одновременно
for _, u := range urls {
    g.Go(func() error { return fetch(ctx, u) })
}
if err := g.Wait(); err != nil { /* первая ошибка */ }

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

Семафор

Ограничить параллелизм без пула: буферизованный канал как счётчик.

sem := make(chan struct{}, 5)
for _, t := range tasks {
    sem <- struct{}{} // занять слот
    go func() {
        defer func() { <-sem }() // освободить
        do(t)
    }()
}

Есть и golang.org/x/sync/semaphore со взвешенным захватом и поддержкой ctx.

Rate limit

time.Ticker даёт простой лимит «не чаще раза в N мс». Для токен-бакета с всплесками есть golang.org/x/time/rate: limiter.Wait(ctx).

Graceful shutdown

Поймать сигнал, перестать принимать новую работу, дождаться текущей с таймаутом:

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
<-ctx.Done()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
srv.Shutdown(shutdownCtx)

Другие

• Generator: функция возвращает канал и сама его заполняет.
• Or-done: обёртка над чтением, которая прекращается по done.
• Single flight (x/sync/singleflight): одновременные запросы одного ключа выполняют работу один раз. Спасает от лавины на кеш.

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

• Канал результатов без закрытия: range по нему висит вечно.
• Горутина на каждую из миллиона задач без ограничения.
• Ошибка в одной задаче не останавливает остальные.
• append в общий слайс из воркеров без мьютекса.

Что спросят на собесе

• Напишите worker pool с ограничением и сохранением порядка.
• Как сделать merge каналов и кто закрывает выходной канал.
• Чем errgroup лучше WaitGroup.
• Как остановить pipeline, если потребитель ушёл.
• Как сделать graceful shutdown сервиса.