Модуль 6. Конкурентность: горутины и каналы

Цели модуля

После модуля студент сможет:

• Запускать и управлять горутинами
• Использовать каналы для коммуникации между горутинами
• Применять select для мультиплексирования каналов
• Защищать общее состояние мьютексами и atomic
• Координировать горутины через WaitGroup и errgroup
• Распознавать и предотвращать race conditions и deadlocks
• Реализовывать паттерн worker pool

🔥 Инцидент: падение в 03:00, которого не было в тестах

Каждую ночь, ровно в окне ночного отчёта, сервис статистики OrderFlow падал с fatal error: concurrent map writes и уходил в рестарт. Днём — неделя за неделей — зелёные тесты, чистый code review и ни одной ошибки. Дежурный уже знал ритуал: посмотреть в 03:10 логи, убедиться, что под перезапустился, и идти спать дальше.

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

go
var stats = map[string]int{}

func track(batch []string) {
    for _, id := range batch {
        go func() { stats[id]++ }() // гонка: общий map без защиты
    }
}

Ночной джоб запускал десятки горутин, которые одновременно писали в один и тот же map, — а map в Go не безопасен для конкурентной записи. Днём гонку не было видно не потому, что её не было, а потому, что планировщик успевал развести горутины по времени. go test ./... без флага -race такую ошибку не показывает вообще.

В этом модуле разберёмся, как из «работает и иногда падает» сделать «работает детерминированно»: горутины, каналы, select, мьютексы и atomic, WaitGroup и errgroup — и главный инструмент диагностики, go test -race.

6.1 Горутины

Что такое горутина

Горутина - лёгкий поток выполнения, управляемый планировщиком Go. Горутины дешевле потоков ОС:

• Стек горутины: 2-8 KB (динамически растёт)
• Стек потока ОС: 1-8 MB (фиксированный)
• Создание горутины: микросекунды
• Создание потока: миллисекунды

Планировщик Go мультиплексирует тысячи горутин на несколько потоков ОС (модель M:N).

Запуск горутины

Ключевое слово go запускает функцию в отдельной горутине:

go
func main() {
    go doWork()  // запускается в новой горутине
    time.Sleep(time.Second)  // ждём завершения (плохой способ)
}

func doWork() {
    fmt.Println("working...")
}

Анонимные горутины

go
go func() {
    fmt.Println("anonymous goroutine")
}()

ВАЖНО: при передаче переменных в анонимную горутину, создавайте локальную копию:

go
// ПЛОХО: гонка по переменной i
for i := 0; i < 10; i++ {
    go func() {
        fmt.Println(i)  // может вывести 10 для всех
    }()
}

// ПРАВИЛЬНО: локальная копия
for i := 0; i < 10; i++ {
    i := i  // создаём новую переменную
    go func() {
        fmt.Println(i)
    }()
}

// Go 1.22+: переменная цикла автоматически создаётся на каждой итерации
for i := 0; i < 10; i++ {
    go func() {
        fmt.Println(i)  // безопасно в Go 1.22+
    }()
}

Горутины и main

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

go
func main() {
    done := make(chan struct{})

    go func() {
        fmt.Println("working...")
        close(done)
    }()

    <-done  // ждём завершения горутины
    fmt.Println("done")

}

Утечки горутин

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

• Запись в канал без читателя
• Ожидание от контекста который не отменяется
• Блокировка на мьютексе который не освободится

go
// УТЕЧКА: горутина заблокирована на записи в канал
func leak() {
    ch := make(chan int)
    go func() {
        ch <- 42  // заблокирована навсегда, нет читателя
    }()
    // выходим из функции, горутина осталась
}

6.2 Каналы

Что такое канал

Канал - типизированный conduit для передачи данных между горутинами. Каналы обеспечивают безопасную коммуникацию без явных блокировок.

"Don't communicate by sharing memory; share memory by communicating."

Создание каналов

go
// Unbuffered канал (ёмкость 0)
ch := make(chan int)

// Buffered канал (ёмкость 10)
ch := make(chan int, 10)

// Канал только для отправки
var sendOnly chan<- int

// Канал только для получения
var receiveOnly <-chan int

Unbuffered каналы

Unbuffered канал блокирует отправителя до готовности получателя. Это синхронная передача данных.

go
func main() {
    ch := make(chan int)

    go func() {
        ch <- 42  // блокируется до чтения
    }()

    value := <-ch  // разблокирует отправителя
    fmt.Println(value)  // 42
}

Buffered каналы

Buffered канал позволяет отправить без блокировки пока есть место.

go
ch := make(chan int, 2)

ch <- 1  // не блокируется (buffer не полон)
ch <- 2  // не блокируется (buffer не полон)
ch <- 3  // БЛОКИРУЕТСЯ (buffer полон, нет читателя)

Отправка и получение

go
ch <- value   // отправка
value := <-ch // получение
value, ok := <-ch  // ok = false если канал закрыт и пуст

Закрытие каналов

text
close(ch)

Правила закрытия:

• Закрывает только отправитель
• Получатель проверяет ok для определения закрытия
• Запись в закрытый канал вызывает panic
• Чтение из закрытого канала вернёт zero value

go
ch := make(chan int)
go func() {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)
}()

for value := range ch {  // range завершится когда канал закрыт
    fmt.Println(value)
}

Паттерн: generator

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

// Использование
for value := range generate(1, 2, 3, 4, 5) {
    fmt.Println(value)
}

Паттерн: pipeline

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

func filter(in <-chan int, predicate func(int) bool) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if predicate(n) {
                out <- n
            }
        }
    }()
    return out
}

// Использование
nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
evens := filter(nums, func(n int) bool { return n%2 == 0 })
squared := square(evens)

for result := range squared {
    fmt.Println(result)  // 4, 16, 36, 64, 100
}

Паттерн: fan-out / fan-in

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

go
func fanIn(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

6.3 select

Базовый синтаксис

select ожидает готовности нескольких каналов и выполняет первый готовый case.

go
select {
case msg := <-ch1:
    fmt.Println("received from ch1:", msg)
case msg := <-ch2:
    fmt.Println("received from ch2:", msg)
case ch3 <- 42:
    fmt.Println("sent to ch3")
default:
    fmt.Println("no channel ready")
}

Правила select

• Если готовы несколько case, выбирается случайный
• default выполняется если ни один канал не готов
• select без default блокирует до готовности канала

Паттерн: таймаут

go
select {
case result := <-longOperation():
    fmt.Println("result:", result)
case <-time.After(5 * time.Second):
    fmt.Println("timeout!")
}

Паттерн: отмена через done-канал

go
func worker(done <-chan struct{}) {
    for {
        select {
        case <-done:
            fmt.Println("cancelled")
            return
        default:
            // выполняем работу
        }
    }
}

func main() {
    done := make(chan struct{})
    go worker(done)

    time.Sleep(3 * time.Second)
    close(done)  // отменяем воркер
    time.Sleep(time.Second)
}

Паттерн: rate limiting

go
func rateLimited(jobs <-chan int, rate time.Duration) {
    ticker := time.NewTicker(rate)
    defer ticker.Stop()

    for job := range jobs {
        <-ticker.C  // ждём тик
        processJob(job)
    }
}

6.4 Мьютексы и sync/atomic

Когда использовать мьютексы

Мьютексы нужны когда горутины изменяют общее состояние. Каналы лучше для передачи данных, мьютексы для защиты состояния.

sync.Mutex

go
type SafeCounter struct {
    mu    sync.Mutex
    count int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}

func (c *SafeCounter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

sync.RWMutex

RWMutex позволяет несколько читателей или одного писателя.

go
type Cache struct {
    mu    sync.RWMutex
    items map[string]string
}

func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    value, ok := c.items[key]
    return value, ok
}

func (c *Cache) Set(key, value string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.items[key] = value
}

sync/atomic

Для простых счётчиков и флагов atomic быстрее мьютексов.

go
import "sync/atomic"

var counter int64

func increment() {
    atomic.AddInt64(&counter, 1)
}

func getValue() int64 {
    return atomic.LoadInt64(&counter)
}

Сравнение подходов

6.5 WaitGroup и errgroup

sync.WaitGroup

WaitGroup ждёт завершения группы горутин.

go
func main() {
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("worker %d\n", id)
        }(i)
    }

    wg.Wait()  // ждём все горутины
    fmt.Println("all done")
}

golang.org/x/sync/errgroup

errgroup - расширение WaitGroup с обработкой ошибок.

go
import "golang.org/x/sync/errgroup"

func fetchAll(ctx context.Context, urls []string) ([]string, error) {
    g, ctx := errgroup.WithContext(ctx)
    results := make([]string, len(urls))
    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            result, err := fetch(ctx, url)
            if err != nil {
                return fmt.Errorf("fetch %s: %w", url, err)
            }
            results[i] = result
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

Преимущества errgroup:

• Автоматическая отмена контекста при первой ошибке
• Ограничение параллелизма через SetLimit
• Чистый API для обработки ошибок

Ограничение параллелизма

go
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(5)  // максимум 5 горутин одновременно

for _, item := range items {
    item := item
    g.Go(func() error {
        return process(ctx, item)
    })
}

return g.Wait()

6.6 Типичные ошибки конкурентности

Race condition

Две горутины одновременно обращаются к общей памяти без синхронизации, и хотя бы одна запись.

go
// ГОНКА
var count int
go func() { count++ }()
count++

Исправление: мьютекс или atomic.

Deadlock

Все горутины заблокированы и не могут продолжить.

go
// DEADLOCK: main ждёт канал, горутина тоже ждёт
ch := make(chan int)
go func() {
    ch <- <-ch  // ждёт сам себя
}()
<-ch

Исправление: правильная последовательность операций, таймауты.

Канал без читателя

go
// УТЕЧКА: горутина заблокирована на записи
ch := make(chan int)
go func() {
    ch <- 42
}()
// никто не читает из ch

Исправление: buffered канал или гарантировать читателя.

Запись в закрытый канал

go
ch := make(chan int)
close(ch)
ch <- 42  // PANIC: send on closed channel

Исправление: закрывает только отправитель, один раз.

Забытый defer wg.Done()

go
var wg sync.WaitGroup
wg.Add(1)
go func() {
    // забыли defer wg.Done()
    process()
}()
wg.Wait()  // заблокируется навсегда

Исправление: defer wg.Done() первой строкой горутины.

Отмена контекста

Всегда проверяйте ctx.Done() в долгих операциях:

go
func worker(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return  // отмена
        default:
            // работа
        }
    }
}

Паттерн: Worker Pool

Worker pool ограничивает количество параллельных воркеров.

go
func workerPool(ctx context.Context, jobs []Job, workers int) []Result {
    jobsCh := make(chan Job)
    resultsCh := make(chan Result)
    var wg sync.WaitGroup

    // Запуск воркеров
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobsCh {
                result := process(ctx, job)
                resultsCh <- result
            }
        }(i)
    }

    // Отправка jobs
    go func() {
        defer close(jobsCh)
        for _, job := range jobs {
            jobsCh <- job
        }
    }()

    // Закрытие results когда все воркеры завершены
    go func() {
        wg.Wait()
        close(resultsCh)
    }()

    // Сбор результатов
    var results []Result
    for result := range resultsCh {
        results = append(results, result)
    }
    return results

}

Привязка к шаблону проекта

Где конкурентность в шаблоне

text
internal/handlers/jobs/
├── outbox_relay/handler.go       # Фоновая обработка
├── saga_orchestrator/handler.go  # Оркестрация саг
└── hmac_rotator/handler.go       # Ротация ключей

internal/adapters/scheduler/
├── cron.go                       # Планировщик
└── job.go                        # Интерфейс Job

Что изучаем на примере шаблона

1. Jobs запускаются по расписанию через cron
2. Каждый job может обрабатывать задачи параллельно
3. errgroup используется для координации
4. Context передаётся для отмены
5. Graceful shutdown через context

Задание: изучите internal/handlers/jobs/

• Как jobs обрабатывают ошибки?
• Есть ли ограничение параллелизма?
• Как реализуется graceful shutdown?
• Используется ли errgroup?

💱 Компромиссы

• Канал vs мьютекс: канал передаёт владение и координирует горутины, но для защиты общей мапы — оверхед и лишние горутины; мьютекс прямее и дешевле для состояния, но легко забыть Lock — и получится гонка как на stats[id]++.
• atomic vs sync.Mutex: atomic.AddInt64 быстрее мьютекса на одиночном счётчике, но защищает ровно одну переменную — мапу так не накрыть; Mutex универсален для любого состояния, но платим блокировкой и риском deadlock.
• Buffered vs unbuffered канал: buffered не блокирует отправителя, пока есть место в буфере, но теряется гарантия «получатель уже принял»; unbuffered синхронен как рукопожатие, но запись без читателя блокирует горутину навсегда — утечка.
• errgroup vs WaitGroup: errgroup сам отменяет контекст при первой ошибке и умеет SetLimit, но это зависимость golang.org/x/sync; WaitGroup лежит в stdlib и предельно прост, но ошибки собираем вручную, а ограничение параллелизма строим сами.
• Worker pool vs горутина на задачу: пул фиксированного размера сглаживает пики и бережёт БД, но добавляет каналы jobs/results и код координации; горутина на каждую задачу пишется одной строкой go, но без потолка параллелизма расход памяти и соединений не ограничен.

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

Задание 1: Простая горутина

Напишите программу которая запускает 10 горутин, каждая печатает свой номер, и ждёт завершения всех.

go
func main() {
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("worker %d\n", id)
        }(i)
    }
    wg.Wait()
}

Задание 2: Pipeline

Реализуйте pipeline: generate -> filter -> square -> print.

go
func generate(nums ...int) <-chan int { ... }
func filter(in <-chan int, fn func(int) bool) <-chan int { ... }
func square(in <-chan int) <-chan int { ... }

Задание 3: Worker Pool

Реализуйте worker pool для обработки задач:

go
type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID int
    Output string
    Err   error
}

func ProcessJobs(ctx context.Context, jobs []Job, workers int) []Result { ... }

Задание 4: Rate Limiter

Реализуйте rate limiter используя ticker:

go
func RateLimitedProcess(ctx context.Context, items []string, rate time.Duration) error {
    ticker := time.NewTicker(rate)
    defer ticker.Stop()

    for _, item := range items {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            process(item)
        }
    }
    return nil
}

Задание 5: Cache с RWMutex

Реализуйте потокобезопасный кэш:

go
type Cache struct {
    mu    sync.RWMutex
    items map[string]string
    ttl   time.Duration
}

func NewCache(ttl time.Duration) *Cache { ... }
func (c *Cache) Get(key string) (string, bool) { ... }
func (c *Cache) Set(key, value string) { ... }
func (c *Cache) Delete(key string) { ... }

Задание 6: errgroup

Реализуйте параллельную загрузку данных с errgroup:

go
func FetchAll(ctx context.Context, urls []string, limit int) ([]string, error) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(limit)
    results := make([]string, len(urls))

    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            data, err := fetch(ctx, url)
            if err != nil {
                return fmt.Errorf("fetch %s: %w", url, err)
            }
            results[i] = data
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil

}

Задание 7: Race detector

Запустите все тесты с race detector:

bash
go test -race ./...

Исправьте все найденные гонки.

Quiz "Проверь себя"

ch := make(chan int, 1)
close(ch)
v, ok := <-ch
fmt.Println(v, ok)

✅ Итоги модуля

Студент должен:

• Запускать и координировать горутины
• Использовать каналы для коммуникации
• Применять select для мультиплексирования
• Защищать состояние мьютексами и atomic
• Использовать WaitGroup и errgroup
• Реализовать worker pool
• Распознавать и предотвращать race conditions
• Реализовать pipeline и fan-in/fan-out
• Ответить на все вопросы Quiz

🔗 Мостик к Модулю 7

Следующий модуль: Модуль 7. Context и отмена операций - context.Context, WithTimeout, WithCancel, WithValue, propagation, graceful shutdown. Привязка к cmd/app/commands/serve.go шаблона.