Go сложный

Паттерны конкурентности, практика руками и три класса проблем

concurrencygoroutines

Вторая часть конспекта по конкурентности — типовые паттерны построения конкурентного кода (worker pool, pipeline, fan-in/fan-out) с рабочим кодом на каждый, формат живой практики "напиши руками за 15-20 минут" (те же паттерны + балансировщик), и три класса проблем, которые в них легко допустить.

Паттерны: worker pool, pipeline, fan-in/fan-out

Worker pool — фиксированное число N горутин читают задачи из общего канала параллельно. Ограничивает степень параллелизма и переиспользует горутины вместо запуска новой на каждую задачу.

Внутри воркера нужны ДВА select (не один!) — один на получение задачи (с ctx.Done()), второй на отправку результата (тоже с ctx.Done()): без второго воркер зависнет навсегда на блокирующей записи результата, если читатель результатов уже ушёл из-за отмены. Частая ошибка: закрыть канал результатов сразу после запуска воркеров, а не после того, как ВСЕ они реально закончили (wg.Wait() в отдельной горутине перед close) — иначе паника send on closed channel у воркера, который ещё пишет.

package main

import (
	"context"
	"fmt"
	"sync"
)

func worker(ctx context.Context, jobs <-chan int, results chan<- int) {
	for {
		select {
		case job, ok := <-jobs:
			if !ok {
				return // канал задач закрыт и вычитан — воркеру больше нечего делать
			}
			select {
			case results <- job * job:
			case <-ctx.Done(): // второй select: читатель результатов мог уже уйти
				return
			}
		case <-ctx.Done():
			return
		}
	}
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	jobs := make(chan int)
	results := make(chan int)

	var wg sync.WaitGroup
	for w := 1; w <= 3; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			worker(ctx, jobs, results)
		}()
	}

	go func() {
		for i := 1; i <= 5; i++ {
			jobs <- i
		}
		close(jobs)
	}()

	go func() {
		wg.Wait() // ждём, пока ВСЕ воркеры реально закончат
		close(results)
	}()

	sum := 0
	for r := range results {
		sum += r
	}
	fmt.Println("сумма квадратов 1..5 =", sum) // 55
}

Pipeline — цепочка стадий, каждая своя горутина (или пул), соединённых каналами: выход одной — вход следующей. Данные текут потоково, не дожидаясь обработки ВСЕХ элементов на предыдущей стадии — снижает пиковую память.

package main

import (
	"context"
	"fmt"
)

func generate(ctx context.Context, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case out <- n * n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	stage1 := generate(ctx, 1, 2, 3, 4) // источник
	stage2 := square(ctx, stage1)       // вторая стадия читает из первой

	for v := range stage2 {
		fmt.Println(v) // 1 4 9 16 — приходят по одному, не разом
	}
}

Fan-out — один источник задач раздаётся нескольким обработчикам (то же самое, что делает worker pool выше: несколько воркеров читают из одного канала jobs). Fan-in — обратное: результаты нескольких каналов сливаются в один выходной, обычно после fan-out, чтобы собрать параллельную работу обратно вместе.

package main

import (
	"fmt"
	"sync"
)

func fanIn(channels ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	wg.Add(len(channels))
	for _, ch := range channels {
		go func(c <-chan int) {
			defer wg.Done()
			for v := range c {
				out <- v
			}
		}(ch)
	}
	go func() {
		wg.Wait() // и здесь — закрываем ТОЛЬКО когда все источники вычитаны
		close(out)
	}()
	return out
}

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

func main() {
	merged := fanIn(gen(1, 2, 3), gen(10, 20, 30))
	sum := 0
	for v := range merged {
		sum += v
	}
	fmt.Println("сумма из двух источников =", sum) // 66
}

Практика: worker pool, балансировщик, pipeline руками

Формат: чистый лист, 15–20 минут, без подсказок — ровно как на секции по Go в Avito/Ozon/Яндексе. Разница между «понимаю каналы» и «умею писать конкурентный код» проверяется именно так — worker pool и pipeline руками смотри выше, здесь третий типовой формат той же практики: балансировщик.

В балансировщике с round-robin+retry под давлением времени легко забыть про потокобезопасность индекса текущего backend'а (нужен Mutex или atomic-счётчик), и про то, что нужно возвращать итоговую ошибку, если ВСЕ backend'ы отказали, а не зависать/молчать.

package main

import (
	"errors"
	"fmt"
	"sync/atomic"
)

type Backend struct {
	name string
	fail bool
}

func call(b Backend) (string, error) {
	if b.fail {
		return "", fmt.Errorf("%s: недоступен", b.name)
	}
	return b.name + ": ok", nil
}

type Balancer struct {
	backends []Backend
	next     atomic.Uint64 // потокобезопасный индекс без Mutex
}

func (b *Balancer) Do() (string, error) {
	n := len(b.backends)
	var lastErr error
	for i := 0; i < n; i++ {
		idx := int(b.next.Add(1)-1) % n
		res, err := call(b.backends[idx])
		if err == nil {
			return res, nil
		}
		lastErr = err // не выходим на первой ошибке — пробуем следующий backend
	}
	return "", errors.Join(errors.New("все backend'ы отказали"), lastErr)
}

func main() {
	b := &Balancer{backends: []Backend{
		{"b1", true}, {"b2", true}, {"b3", false},
	}}
	fmt.Println(b.Do()) // два отказа, третий отвечает — b3: ok <nil>

	allDown := &Balancer{backends: []Backend{{"b1", true}, {"b2", true}}}
	fmt.Println(allDown.Do()) // все отказали — явная итоговая ошибка, не зависание
}

В pipeline (см. код выше) — классическая ловушка: читать входной канал через for chunk := range chunks вместо select с ctx.Done() — range не видит отмену контекста и не разблокируется, пока канал не закроют.

Интервьюер почти всегда спрашивает ПОСЛЕ того, как код написан: «а что если один backend не ошибку вернёт, а просто зависнет навсегда?» — прямая проверка, обёрнут ли вызов backend в select с ctx.Done(), а не просто вызван напрямую.

Утечка горутин, deadlock, race condition — три разных класса проблем

На собеседовании важно называть точный термин, а не «что-то не так с горутинами» — это три РАЗНЫХ класса проблем, которые часто путают между собой.

Утечка горутины — горутина заблокирована навсегда (обычно на отправке в канал, который никто не читает). Утечки накапливаются, каждая держит стек и ресурсы, пока процесс не упадёт по OOM. Профилактика: у КАЖДОЙ горутины должен быть явный гарантированный способ завершиться.

Deadlock — две+ горутины взаимно ждут друг друга, ПОЛНАЯ остановка. Классическая причина — захват нескольких мьютексов в разном порядке разными горутинами (ABBA-deadlock). Go детектирует deadlock, только если ВСЕ горутины программы заблокированы одновременно (паника all goroutines are asleep) — частичный deadlock (часть горутин продолжает работать) рантайм не ловит.

Race condition — конкурентный доступ к памяти без синхронизации, где хотя бы одно обращение — запись. counter++ — это чтение+инкремент+запись, не одна атомарная операция; результат непредсказуем, но программа НЕ падает и НЕ виснет (в отличие от двух пунктов выше) — просто даёт неверный ответ.

Ключевой факт про инструменты: go run -race находит гонки данных, но НЕ находит deadlock — это разные классы проблем с разной диагностикой. Конкурентная запись в обычную (не sync.Map) map без синхронизации — не тихая гонка, а немедленная паника рантайма fatal error: concurrent map writes, в отличие от гонки на простом int-счётчике, где часть инкрементов просто молча теряется.

Самопроверка 0 / 4

Помню, зачем воркеру в worker pool нужны ДВА select, а не один

Могу чётко различить утечку горутины, deadlock и race condition — это три разных класса проблем

Знаю, что go run -race находит гонки, но НЕ находит deadlock

Помню, что конкурентная запись в обычную map — не тихая гонка, а немедленная паника fatal error

Как усвоено?

← Назад Горутины, каналы, select Дальше → Пакет sync: Mutex, WaitGroup, Once, Map, Pool, Cond, errgroup, atomic