Горутины - легковесная конкурентность

Горутины - легковесная конкурентность

Горутины - это killer feature Go. Это легковесные потоки выполнения, управляемые runtime Go. Можно запустить тысячи горутин без проблем с производительностью.

Основы горутин

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

func sayHello(name string) {
    fmt.Printf("Привет, %s!\n", name)
}

func main() {
    // Обычный вызов
    sayHello("Мир")

    // Запуск в горутине
    go sayHello("Горутина")

    // Без этого main завершится раньше горутины
    time.Sleep(time.Second)
}

Почему горутин могут быть тысячи

Каждая горутина стартует с маленьким стеком - всего 2 КБ. Это в сотни раз меньше OS-потока. Когда стек переполняется, runtime копирует его в новый блок большего размера и поправляет указатели.

Поэтому 1000 горутин обходится примерно в 2 МБ - на два-три порядка дешевле такого же числа потоков ОС.

Отсюда не следует, что горутины бесплатны. По той же арифметике миллион горутин - это уже около 2 ГБ только под стеки, и это при минимальном стеке: он растёт по мере вложенности вызовов. Добавьте к этому работу планировщика и давление на GC, которому все эти стеки надо просматривать. Ниже в уроке есть таблица, где «миллионы горутин» стоят в колонке «плохо», и это не противоречие - дешёвая горутина и неограниченное их количество разные вещи. Когда задач много, число одновременных горутин ограничивают: worker pool или семафор.

Анонимные функции

Подробнее про анонимные функции и замыкания - в уроке про функции.

func main() {
    message := "Hello"

    go func() {
        fmt.Println(message) // захват переменной
    }()

    go func(msg string) {
        fmt.Println(msg) // передача параметром (безопаснее)
    }(message)

    time.Sleep(time.Second)
}

Синхронизация с WaitGroup

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // уменьшаем счетчик при завершении

    fmt.Printf("Worker %d начал работу\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %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("Все workers завершены")
}

Race conditions и синхронизация

Проблема race condition

// ПЛОХО: race condition
var counter int

func increment() {
    counter++ // не атомарная операция!
}

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            increment()
        }()
    }

    wg.Wait()
    fmt.Println(counter) // не всегда 1000!
}

Решение с Mutex

var (
    counter int
    mu      sync.Mutex
)

func increment() {
    mu.Lock()
    defer mu.Unlock()
    counter++
}

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

var counter int64

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

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

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

Когда базовых горутин и WaitGroup уже недостаточно, в дело вступают комбинации горутин с каналами: worker pool, fan-in / fan-out, pipeline, rate limiting, errgroup. Это уже не про синтаксис, а про композицию - поэтому их разбор вынесен в отдельный трек:

• Паттерны: fan-in, fan-out, pipeline, worker pool
• errgroup и graceful shutdown
• Race conditions и тестирование конкурентного кода
• Горутины под капотом: планировщик GMP

В этом уроке наша цель - понять, как горутины устроены и синхронизируются. Применение к реальным задачам - следующий шаг.

Контекст отмены

func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d остановлен\n", id)
            return
        default:
            // выполняем работу
            fmt.Printf("Worker %d работает\n", id)
            time.Sleep(time.Second)
        }
    }
}

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

    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }

    time.Sleep(3 * time.Second)
    cancel() // останавливаем всех

    time.Sleep(time.Second)
}

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

Захват переменной цикла

// ОШИБКА - но только до Go 1.22.
// На современной версии этот код печатает 0..4: см. врезку ниже.
for i := 0; i < 5; i++ {
    go func() {
        fmt.Println(i) // до 1.22: все горутины видели i = 5
    }()
}

// Работает на любой версии: значение передаётся аргументом
for i := 0; i < 5; i++ {
    go func(n int) {
        fmt.Println(n)
    }(i) // передаём копию
}

// Идиома ДО Go 1.22: локальная копия внутри тела цикла.
// С 1.22 эта строка ничего не меняет - переменная и так новая.
for i := 0; i < 5; i++ {
    i := i
    go func() {
        fmt.Println(i)
    }()
}

// ПЛОХО: горутина никогда не завершится
func leak() {
    ch := make(chan int)

    go func() {
        val := <-ch // блокируется навсегда
        fmt.Println(val)
    }()

    // канал не закрыт, никто не пишет
}

// ХОРОШО: используем контекст
func noLeak(ctx context.Context) {
    ch := make(chan int)

    go func() {
        select {
        case val := <-ch:
            fmt.Println(val)
        case <-ctx.Done():
            return
        }
    }()
}

// Количество горутин
fmt.Println("Горутин:", runtime.NumGoroutine())

// Профилирование
import _ "net/http/pprof"

func main() {
    go func() {
        log.Println(http.ListenAndServe("localhost:6060", nil))
    }()

    // основная логика
}

// Затем: go tool pprof http://localhost:6060/debug/pprof/goroutine

type Result struct {
    URL   string
    Links []string
    Err   error
}

func crawl(url string) Result {
    // имитация работы
    time.Sleep(100 * time.Millisecond)

    if rand.Float32() < 0.1 {
        return Result{URL: url, Err: errors.New("failed to fetch")}
    }

    // генерируем случайные ссылки
    numLinks := rand.Intn(5)
    links := make([]string, numLinks)
    for i := range links {
        links[i] = fmt.Sprintf("%s/%d", url, i)
    }

    return Result{URL: url, Links: links}
}

func crawler(ctx context.Context, startURL string, maxDepth int) {
    visited := make(map[string]bool)
    var mu sync.Mutex

    var wg sync.WaitGroup
    results := make(chan Result)

    // Запускаем сборщик результатов
    go func() {
        for result := range results {
            if result.Err != nil {
                fmt.Printf("Ошибка %s: %v\n", result.URL, result.Err)
                continue
            }
            fmt.Printf("Обработано %s: %d ссылок\n", result.URL, len(result.Links))
        }
    }()

    var crawlURL func(string, int)
    crawlURL = func(url string, depth int) {
        if depth > maxDepth {
            return
        }

        mu.Lock()
        if visited[url] {
            mu.Unlock()
            return
        }
        visited[url] = true
        mu.Unlock()

        wg.Add(1)
        go func() {
            defer wg.Done()

            select {
            case <-ctx.Done():
                return
            default:
                result := crawl(url)
                results <- result

                for _, link := range result.Links {
                    crawlURL(link, depth+1)
                }
            }
        }()
    }

    crawlURL(startURL, 0)
    wg.Wait()
    close(results)
}