Как в Go управлять параллелизмом и ограничивать количество одновременно выполняемых горутин?

Question

Как в Go управлять параллелизмом и ограничивать количество одновременно выполняемых горутин?

Тип вопроса: Golang Вероятность: 10% Категория: Golang Разработчик

Ответ

В Go управление параллелизмом осуществляется не через прямое управление потоками ОС, а через горутины — легковесные потоки, управляемые рантаймом Go.

GOMAXPROCS

Переменная runtime.GOMAXPROCS определяет, сколько потоков ОС могут одновременно исполнять код Go. По умолчанию это значение равно количеству доступных ядер ЦП. Это не ограничивает число горутин, а лишь задает степень реального параллелизма.

// Устанавливает количество потоков ОС равным количеству ядер
runtime.GOMAXPROCS(runtime.NumCPU())

Способы ограничения количества горутин

Для ограничения количества одновременно работающих горутин (конкарренси) используются следующие паттерны:

Пул воркеров (Worker Pool): Создается фиксированное количество горутин-воркеров, которые разбирают задачи из общего канала.

tasks := make(chan int, 100)
var wg sync.WaitGroup

// Запускаем 5 воркеров
for i := 0; i < 5; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for task := range tasks {
            // Обработка задачи
        }
    }()
}

// ... добавляем задачи в канал tasks ...
close(tasks)
wg.Wait() // Ждем завершения всех воркеров

Канал-семафор: Канал с буфером используется для ограничения входа в критическую секцию.

var maxConcurrency = 5
sem := make(chan struct{}, maxConcurrency)

for _, task := range tasks {
    sem <- struct{}{} // Занимаем слот
    go func(t Task) {
        defer func() { <-sem }() // Освобождаем слот
        process(t)
    }(task)
}

Семафор из пакета sync: Более современный и идиоматичный способ, доступный в golang.org/x/sync/semaphore.

import "golang.org/x/sync/semaphore"

var ( 
    maxWorkers = int64(5)
    sem = semaphore.NewWeighted(maxWorkers)
    ctx = context.Background()
)

for _, task := range tasks {
    if err := sem.Acquire(ctx, 1); err != nil {
        // Обработка ошибки
        break
    }
    go func(t Task) {
        defer sem.Release(1)
        process(t)
    }(task)
}

Важно: Несмотря на легковесность горутин, их неограниченное создание для ресурсоемких операций (например, сетевые запросы, работа с БД) может привести к исчерпанию файловых дескрипторов, памяти или других системных ресурсов.

Ответ 18+ 🔞

Так, слушай, про параллелизм в Go. Тут всё не как у людей, блядь. Нет тебе этих потоков ОС, с которыми вечно морока. Вместо них — горутины, такие легковесные потоки, которыми рантайм сам управляет, как хочет. Проще пареной репы, ёпта.

GOMAXPROCS — это не про лимит горутин!

Первое, что все путают — runtime.GOMAXPROCS. Это, сука, не лимит на количество горутин, нет! Это просто число потоков ОС, которые могут одновременно код Go исполнять. По умолчанию — равно количеству ядер на твоём процессоре. Хоть миллион горутин создавай — они будут по этим потокам распределяться.

// Выставим потоки ОС по числу ядер — обычно так и оставляют
runtime.GOMAXPROCS(runtime.NumCPU())

А как тогда ограничить, чтобы всё не легло?

А вот это уже вопрос правильный. Горутины-то легкие, но если ты на каждую тапку будешь свою горутину создавать под тяжёлую операцию (типа запроса в базу или к внешнему API), то упрёшься в лимиты системы — дескрипторы, память, сетевые сокеты. Пиздец, а не работа. Поэтому есть проверенные паттерны:

Пул воркеров (Worker Pool) Классика, блядь. Создаёшь стабильную банду горутин-работяг, которые сидят и ждут задач из общего канала.

tasks := make(chan int, 100)
var wg sync.WaitGroup

// Запускаем, например, 5 работяг
for i := 0; i < 5; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for task := range tasks {
            // Тут делаем полезную хуйню
        }
    }()
}

// ... кидаем задачи в канал tasks ...
close(tasks) // Говорим, что всё, задач больше не будет
wg.Wait() // Ждём, пока последний ленивец работу закончит

Канал-семафор Хитрый способ, вроде как идиоматичный для Go. Создаёшь канал с буфером, скажем, на 5 мест. Прежде чем запустить горутину, пытаешься запихнуть в него пустую структуру — если места нет, то ждёшь. Горутина, завершаясь, вытаскивает эту структуру обратно, освобождая слот.
```
var maxConcurrency = 5
sem := make(chan struct{}, maxConcurrency)

for _, task := range tasks {
    sem <- struct{}{} // Захватываем слот. Если мест нет — встанем тут.
    go func(t Task) {
        defer func() { <-sem }() // Освобождаем слот, когда всё сделали
        process(t)
    }(task)
}
```

Нормальный семафор из sync/semaphore Самый, блядь, цивилизованный способ, хотя и требует внешнего пакета (golang.org/x/sync/semaphore). Создаёшь семафор с определённым «весом» и просто Acquire/Release делаешь.

import "golang.org/x/sync/semaphore"

var ( 
    maxWorkers = int64(5)
    sem = semaphore.NewWeighted(maxWorkers)
    ctx = context.Background()
)

for _, task := range tasks {
    if err := sem.Acquire(ctx, 1); err != nil {
        // Что-то пошло не так, сдаёмся
        break
    }
    go func(t Task) {
        defer sem.Release(1)
        process(t)
    }(task)
}

Главное запомнить: сам Go тебя от дурости не спасёт. Можно нагородить горутин — овердохуища, а потом сидеть и думать, почему сервис лег от нехватки ресурсов. Ограничивай их, особенно для I/O операций, будь умником, а не распиздяем.