Горутины и каналы в Go: конкурентность для начинающих

Горутины — это то, за что любят Go. Представь официанта в ресторане: он не ждёт пока один столик доест суп, чтобы принять заказ у другого — он обслуживает несколько столиков одновременно. Горутины в Go работают так же: лёгкие потоки выполнения, которые позволяют программе делать много дел «одновременно». И это встроено прямо в язык!

Горутины: лёгкие потоки

package main

import (
    "fmt"
    "time"
)

func sayHello(name string) {
    for i := 0; i < 3; i++ {
        fmt.Printf("Привет от %s! (%d)\n", name, i)
        time.Sleep(100 * time.Millisecond)
    }
}

func main() {
    // Обычный вызов — ждём завершения
    sayHello("Алиса")
    
    // Горутина — запускаем в фоне
    go sayHello("Борис")  // go запускает функцию как горутину
    go sayHello("Вера")   // Ещё одна горутина
    
    fmt.Println("main продолжает работу!")
    
    // Проблема: main может завершиться до горутин!
    time.Sleep(500 * time.Millisecond) // Ждём (не лучший способ)
    fmt.Println("Конец")
}

WaitGroup: правильное ожидание горутин

package main

import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // Сообщаем что горутина завершена
    
    fmt.Printf("Воркер %d начал работу\n", id)
    time.Sleep(time.Duration(id) * 100 * time.Millisecond)
    fmt.Printf("Воркер %d закончил\n", id)
}

func main() {
    var wg sync.WaitGroup // Счётчик активных горутин
    
    for i := 1; i <= 5; i++ {
        wg.Add(1)           // Увеличиваем счётчик
        go worker(i, &wg)   // Запускаем горутину
    }
    
    wg.Wait() // Ждём пока счётчик не станет 0
    fmt.Println("Все воркеры завершили работу!")
}

Каналы: трубы для передачи данных

Канал (channel) — это труба, по которой горутины передают данные друг другу. Представь пневматическую почту в банке: отправляешь капсулу — получатель принимает её на другом конце.

package main

import "fmt"

func main() {
    // Создание канала
    ch := make(chan int) // Канал для передачи int
    
    // Отправляем значение в горутине
    go func() {
        ch <- 42 // Отправка в канал (стрелка влево)
    }()
    
    // Получаем значение
    value := <-ch // Получение из канала (стрелка влево = из канала)
    fmt.Println(value) // 42
    
    // Двусторонний обмен
    done := make(chan bool)
    
    go func() {
        fmt.Println("Горутина работает...")
        done <- true // Сигнал о завершении
    }()
    
    <-done // Ждём сигнала
    fmt.Println("Горутина завершена")
}

Буферизованные каналы

package main

import "fmt"

func main() {
    // Небуферизованный — блокирует пока не прочитают
    ch1 := make(chan int)
    
    // Буферизованный — не блокирует пока буфер не заполнен
    ch2 := make(chan int, 3) // буфер на 3 элемента
    
    // Можно отправить без горутины (не блокирует сразу)
    ch2 <- 1
    ch2 <- 2
    ch2 <- 3
    // ch2 <- 4 // ЗАБЛОКИРУЕТ — буфер полон!
    
    fmt.Println(<-ch2) // 1
    fmt.Println(<-ch2) // 2
    fmt.Println(<-ch2) // 3
    
    // Закрытие канала
    jobs := make(chan int, 5)
    
    go func() {
        for j := 1; j <= 5; j++ {
            jobs <- j
        }
        close(jobs) // Закрываем канал — больше отправок не будет
    }()
    
    // range читает пока канал не закрыт
    for job := range jobs {
        fmt.Printf("Получен job: %d\n", job)
    }
}

Паттерн producer-consumer

package main

import (
    "fmt"
    "sync"
)

func producer(ch chan<- int, n int) { // chan<- только для отправки
    for i := 0; i < n; i++ {
        ch <- i * i // Отправляем квадраты чисел
    }
    close(ch)
}

func consumer(id int, ch <-chan int, results chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    for value := range ch { // Читаем пока канал не закрыт
        results <- fmt.Sprintf("Consumer %d: получил %d", id, value)
    }
}

func main() {
    jobs := make(chan int, 10)
    results := make(chan string, 20)
    var wg sync.WaitGroup
    
    // Запускаем 3 потребителя
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go consumer(i, jobs, results, &wg)
    }
    
    // Производитель
    go producer(jobs, 9)
    
    // Закрываем results когда все потребители завершились
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // Выводим результаты
    for r := range results {
        fmt.Println(r)
    }
}

select: множественные каналы

package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "один"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "два"
    }()
    
    // select ждёт первый доступный канал
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println("Получено из ch1:", msg1)
        case msg2 := <-ch2:
            fmt.Println("Получено из ch2:", msg2)
        }
    }
    
    // select с default — неблокирующий
    ch3 := make(chan int, 1)
    select {
    case v := <-ch3:
        fmt.Println("Получено:", v)
    default:
        fmt.Println("Канал пуст — не блокируем")
    }
    
    // select с таймаутом
    ch4 := make(chan string)
    select {
    case result := <-ch4:
        fmt.Println(result)
    case <-time.After(500 * time.Millisecond):
        fmt.Println("Таймаут! Слишком долго ждём")
    }
}

Mutex: защита от race condition

package main

import (
    "fmt"
    "sync"
)

// Race condition — несколько горутин меняют одну переменную
type SafeCounter struct {
    mu    sync.Mutex // Мьютекс — замок
    count int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()   // Захватываем замок
    c.count++
    c.mu.Unlock() // Отпускаем замок
}

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

func main() {
    counter := &SafeCounter{}
    var wg sync.WaitGroup
    
    // 1000 горутин одновременно увеличивают счётчик
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }
    
    wg.Wait()
    fmt.Println("Итоговый счётчик:", counter.Value()) // 1000
    // Без мьютекса результат был бы < 1000 (race condition!)
}

Типичные ошибки

Дедлок — все горутины заблокированы

// ОШИБКА: deadlock!
ch := make(chan int)
ch <- 1  // Блокируется — никто не читает!
v := <-ch

// ПРАВИЛЬНО: отправка в горутине
ch := make(chan int)
go func() { ch <- 1 }()
v := <-ch
fmt.Println(v)

Итого

• ✅ Горутина — лёгкий поток, запускается через go func()
• ✅ sync.WaitGroup — ждём завершения группы горутин
• ✅ Канал (chan) — труба для передачи данных между горутинами
• ✅ Буферизованный канал — make(chan T, n) — не блокирует до заполнения буфера
• ✅ select — ожидание первого доступного из нескольких каналов
• ✅ sync.Mutex — защита от race condition при работе с общими данными

Атомарные операции

package main

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

func main() {
    // Атомарный счётчик — быстрее мьютекса для простых операций
    var counter int64
    var wg sync.WaitGroup
    
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            atomic.AddInt64(&counter, 1)  // Потокобезопасное сложение
        }()
    }
    
    wg.Wait()
    fmt.Println("Счётчик:", atomic.LoadInt64(&counter))  // 1000
}

Паттерн Pipeline

package main

import "fmt"

// Pipeline — цепочка горутин обрабатывающих данные
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums { out <- n }
        close(out)
    }()
    return out
}

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

func filterEven(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            if n%2 == 0 { out <- n }
        }
        close(out)
    }()
    return out
}

func main() {
    // Строим pipeline: числа → квадраты → только чётные
    nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squares := square(nums)
    evenSquares := filterEven(squares)
    
    for n := range evenSquares {
        fmt.Print(n, " ")  // 4 16 36 64 100
    }
    fmt.Println()
}

context: отмена и таймауты

package main

import (
    "context"
    "fmt"
    "time"
)

func fetchData(ctx context.Context, id int) (string, error) {
    select {
    case <-ctx.Done():
        return "", ctx.Err()  // Контекст отменён или таймаут
    case <-time.After(time.Duration(id*100) * time.Millisecond):
        return fmt.Sprintf("Данные для %d", id), nil
    }
}

func main() {
    // Таймаут 250ms
    ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
    defer cancel()
    
    for i := 1; i <= 5; i++ {
        result, err := fetchData(ctx, i)
        if err != nil {
            fmt.Printf("ID %d: ошибка — %v\n", i, err)
        } else {
            fmt.Println(result)
        }
    }
}

sync.Once и sync.Pool

package main

import (
    "fmt"
    "sync"
)

// sync.Once — выполнить функцию ровно один раз
type Singleton struct {
    Data string
}

var instance *Singleton
var once sync.Once

func GetInstance() *Singleton {
    once.Do(func() {
        instance = &Singleton{Data: "Инициализирован!"}
        fmt.Println("Создаём синглтон...")
    })
    return instance
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            s := GetInstance()
            fmt.Println(s.Data)
        }()
    }
    wg.Wait()
    // "Создаём синглтон..." выведется только один раз!
}

Итого

• ✅ Горутина — лёгкий поток: go func()
• ✅ sync.WaitGroup — ждать завершения группы горутин
• ✅ Канал (chan T) — труба для передачи данных
• ✅ Направленные каналы: chan<- (только отправка), <-chan (только получение)
• ✅ Буферизованный канал: make(chan T, n)
• ✅ select — ждать первый доступный канал
• ✅ sync.Mutex — защита от race condition
• ✅ atomic — быстрые атомарные операции
• ✅ context — отмена и таймауты для горутин
• ✅ Паттерн Pipeline — цепочка горутин для обработки данных

HTTP сервер на Go

package main

import (
    "fmt"
    "net/http"
    "sync"
    "time"
)

// Счётчик запросов — thread-safe
type Counter struct {
    mu    sync.Mutex
    count int
}

func (c *Counter) Increment() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
    return c.count
}

func main() {
    counter := &Counter{}
    
    http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
        n := counter.Increment()
        fmt.Fprintf(w, "Запрос #%d\nВремя: %s\n", n, time.Now().Format("15:04:05"))
    })
    
    http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        fmt.Fprintf(w, `{"status":"ok","time":"%s"}`, time.Now().Format(time.RFC3339))
    })
    
    fmt.Println("Сервер запущен на :8080")
    fmt.Println("Открой http://localhost:8080 в браузере")
    http.ListenAndServe(":8080", nil)
}

Паттерн Worker Pool

package main

import (
    "fmt"
    "sync"
    "time"
)

type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID  int
    Output string
    Error  error
}

func processJob(job Job) Result {
    time.Sleep(time.Millisecond * 100) // Симуляция работы
    return Result{
        JobID:  job.ID,
        Output: fmt.Sprintf("Обработан: %s", job.Data),
    }
}

func workerPool(numWorkers int, jobs <-chan Job, results chan<- Result) {
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for job := range jobs {
                result := processJob(job)
                fmt.Printf("  Worker %d обработал job %d\n", workerID, job.ID)
                results <- result
            }
        }(i + 1)
    }
    go func() {
        wg.Wait()
        close(results)
    }()
}

func main() {
    jobsCh := make(chan Job, 10)
    resultsCh := make(chan Result, 10)
    
    go workerPool(3, jobsCh, resultsCh)
    
    // Отправляем задачи
    for i := 1; i <= 9; i++ {
        jobsCh <- Job{ID: i, Data: fmt.Sprintf("данные_%d", i)}
    }
    close(jobsCh)
    
    // Собираем результаты
    for r := range resultsCh {
        fmt.Printf("Результат job %d: %s\n", r.JobID, r.Output)
    }
}

Практика: ограничение конкурентности

package main

import (
    "fmt"
    "sync"
    "time"
)

// Semaphore — ограничение числа одновременных горутин
type Semaphore chan struct{}

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

func (s Semaphore) Acquire() { s <- struct{}{} }
func (s Semaphore) Release() { <-s }

func main() {
    sem := NewSemaphore(3) // Не более 3 одновременных задач
    var wg sync.WaitGroup
    
    for i := 1; i <= 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            sem.Acquire()
            defer sem.Release()
            
            fmt.Printf("Задача %d начата\n", id)
            time.Sleep(200 * time.Millisecond)
            fmt.Printf("Задача %d завершена\n", id)
        }(i)
    }
    
    wg.Wait()
    fmt.Println("Все задачи выполнены!")
}

Дополнительные паттерны конкурентности

package main

import (
    "fmt"
    "sync"
)

// sync.Map — потокобезопасный map
func main() {
    var m sync.Map
    var wg sync.WaitGroup
    
    // Несколько горутин пишут одновременно
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            m.Store(fmt.Sprintf("key%d", id), id*10)
        }(i)
    }
    
    wg.Wait()
    
    // Перебор элементов
    m.Range(func(key, value interface{}) bool {
        fmt.Printf("%s = %d\n", key, value)
        return true // true = продолжить перебор
    })
}