Как в Go создать worker pool с ограничением concurrency?

Как в Go реализовать worker pool с ограниченным concurrency?

• concurrency определяет число одновременно работающих goroutine
• Для ограничения создают буферизированный канал размером с concurrency
• Перед запуском worker в канал помещают «токен»; при заполнении канала выполнение ожидает
• После завершения worker извлекает токен и освобождает слот
• Фиксированный набор worker’ов получает задачи из очереди, представленной каналом
• Канал задач регулирует распределение работы между worker’ами
• Решение помогает контролировать нагрузку и не допускать исчерпания CPU и памяти

Идея: канал с токенами ограничивает число одновременно выполняющихся горутин, управляет concurrency и защищает систему от перегрузки.

Подробный ответ

Основной ответ

Worker pool с ограничением concurrency в Go строится на фиксированном числе воркеров (goroutine), которые читают задачи из общей очереди-канала. Такой дизайн ограничивает количество одновременно активных обработчиков, благодаря чему ресурсы используются предсказуемо и система не перегружается.

Ключевые моменты

• Очередь обычно реализуют с помощью буферизированного или небуферизированного канала: продюсер помещает в него задачи, а воркеры последовательно забирают их и выполняют.
• Максимальный параллелизм задаётся количеством горутин: в цикле запускают ровно N воркеров, где N — допустимое число одновременных операций.
• Чтобы дождаться завершения всех воркеров, применяют sync.WaitGroup. Он не позволяет основной горутине завершиться раньше обработчиков.
• Также необходимо предусмотреть обработку ошибок и отмену. Для graceful shutdown и таймаутов часто передают context.Context.
• Подобную схему используют в продакшене при ограничении обращений к внешним сервисам, выполнении CPU-intensive задач и других подобных операций.

Практический контекст

В проектах на Go 1.18+ worker pool применяют для ограничения запросов к API, параллельной обработки данных и фоновых операций. Например, при сборе метрик с множества источников значение concurrency выбирают с учётом лимитов внешних систем.

Пример кода для иллюстрации:

type Task func()

func worker(id int, tasks <-chan Task, wg *sync.WaitGroup) {
defer wg.Done()
for task := range tasks {
task() // Выполнение задачи
}
}

func main() {
const maxWorkers = 5
tasks := make(chan Task)
var wg sync.WaitGroup

for i := 0; i < maxWorkers; i++ {
wg.Add(1)
go worker(i, tasks, &wg)
}

for i := 0; i < 20; i++ {
taskNum := i
tasks <- func() {
fmt.Printf("Выполняется задача %d\n", taskNum)
time.Sleep(time.Second)
}
}
close(tasks)
wg.Wait()
}

В результате создаются 5 воркеров, которые выполняют 20 задач последовательно, не превышая concurrency 5.