Очереди и воркеры на Go: практическое руководство

Зачем вообще воркер-пул, если есть горутины

Создать горутину на каждый входящий запрос — первый порыв любого, кто открыл Go. На десяти задачах это работает. На десяти тысячах — тоже, но каждая горутина занимает минимум 2–4 КБ стека, а при обращении к внешнему сервису или базе вы упираетесь не в память, а в лимит соединений. Пул воркеров решает конкретную задачу: ограничить параллелизм до разумного числа и не потерять задачи при остановке сервиса.

В этом руководстве — рабочий паттерн на стандартной библиотеке Go 1.22+, без сторонних пакетов. Всё копируется и запускается. Внешние брокеры (RabbitMQ, NATS, Kafka) — отдельная тема, здесь рассматриваем только in-process-очереди на каналах.

Что нужно до начала

• Go 1.22 или новее (проверьте: go version).
• Любой текстовый редактор или IDE с поддержкой Go.
• Терминал, в котором доступна команда go run.

Шаг 1. Минимальный воркер-пул

Создаём файл main.go. Пул состоит из трёх элементов: канал задач, горутины-воркеры и sync.WaitGroup для ожидания завершения.

// main.go — минимальный worker pool, Go 1.22+
package main

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

// Task описывает единицу работы.
type Task struct {
	ID      int
	Payload string
}

// worker читает задачи из канала и «обрабатывает» их.
func worker(id int, tasks <-chan Task, wg *sync.WaitGroup) {
	defer wg.Done()
	for task := range tasks {
		fmt.Printf("воркер %d: обрабатываю задачу %d (%s)\n", id, task.ID, task.Payload)
		time.Sleep(200 * time.Millisecond) // имитация полезной работы
	}
	fmt.Printf("воркер %d: канал закрыт, завершаюсь\n", id)
}

func main() {
	const numWorkers = 3
	const numTasks = 10

	tasks := make(chan Task, numTasks) // буферизированный канал
	var wg sync.WaitGroup

	// Запускаем воркеры
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go worker(i, tasks, &wg)
	}

	// Отправляем задачи
	for j := 1; j <= numTasks; j++ {
		tasks <- Task{ID: j, Payload: fmt.Sprintf("data-%d", j)}
	}
	close(tasks) // сигнал воркерам: новых задач не будет

	wg.Wait() // ждём, пока все воркеры завершатся
	fmt.Println("все задачи обработаны")
}

Проверка: запустите go run main.go. В выводе каждая из десяти задач должна появиться ровно один раз, а последней строкой — «все задачи обработаны».

Почему канал буферизированный

Буфер размером numTasks позволяет продюсеру положить все задачи, не блокируясь. Если задачи приходят потоком (HTTP-запросы, события из файла), разумный буфер — от десятков до сотен элементов. Слишком большой буфер скрывает проблему: воркеры не справляются, а вы узнаёте об этом только по росту потребления памяти.

Шаг 2. Результаты и обработка ошибок

Реальная работа возвращает результат или ошибку. Добавляем канал результатов.

// result.go — воркер с результатами, Go 1.22+
package main

import (
	"errors"
	"fmt"
	"sync"
)

type Job struct {
	ID   int
	Data int
}

type Result struct {
	JobID int
	Value int
	Err   error
}

func process(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()
	for job := range jobs {
		if job.Data < 0 {
			results <- Result{JobID: job.ID, Err: errors.New("отрицательное значение")}
			continue
		}
		results <- Result{JobID: job.ID, Value: job.Data * 2}
	}
}

func main() {
	jobs := make(chan Job, 8)
	results := make(chan Result, 8)
	var wg sync.WaitGroup

	for i := 0; i < 4; i++ {
		wg.Add(1)
		go process(i, jobs, results, &wg)
	}

	// Закрываем results после завершения всех воркеров
	go func() {
		wg.Wait()
		close(results)
	}()

	// Отправляем задачи
	go func() {
		for _, v := range []int{1, 5, -3, 7, 0, 42} {
			jobs <- Job{ID: v, Data: v}
		}
		close(jobs)
	}()

	// Читаем результаты
	for r := range results {
		if r.Err != nil {
			fmt.Printf("задача %d: ошибка — %v\n", r.JobID, r.Err)
		} else {
			fmt.Printf("задача %d: результат %d\n", r.JobID, r.Value)
		}
	}
}

Проверка: задача с отрицательным числом выводит ошибку, остальные — удвоенное значение. Порядок может отличаться — это нормально для параллельного выполнения.

Шаг 3. Graceful shutdown через context

В продакшене сервис получает SIGTERM, и у вас есть секунды, чтобы доработать текущие задачи и не потерять новые. Стандартный context и пакет os/signal решают это без сторонних библиотек.

// shutdown.go — graceful shutdown, Go 1.22+
package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

func gracefulWorker(ctx context.Context, id int, tasks <-chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		select {
		case task, ok := <-tasks:
			if !ok {
				return // канал закрыт, задач больше нет
			}
			fmt.Printf("воркер %d: задача %d\n", id, task)
			time.Sleep(500 * time.Millisecond)
		case <-ctx.Done():
			fmt.Printf("воркер %d: получен сигнал завершения\n", id)
			return
		}
	}
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

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

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

	// Генератор задач: остановится, когда контекст отменён
	go func() {
		for n := 0; ; n++ {
			select {
			case tasks <- n:
			case <-ctx.Done():
				close(tasks)
				return
			}
		}
	}()

	wg.Wait()
	fmt.Println("сервис остановлен корректно")
}

Проверка: запустите программу и нажмите Ctrl+C. Воркеры сообщат о завершении, последней строкой выведется «сервис остановлен корректно». Если строки нет — wg.Wait() не дождался всех воркеров, ищите незакрытый канал.

Канал в Go — это и очередь, и механизм синхронизации. Но именно поэтому ошибка в его использовании бьёт сразу по двум местам: по данным и по порядку завершения.

Типичные ошибки и как их избежать

Когда in-process очереди недостаточно

Каналы Go живут в памяти одного процесса. Если вам нужно одно из перечисленного ниже, смотрите в сторону внешнего брокера:

• Персистентность — задачи не должны теряться при падении процесса.
• Распределённые консьюмеры — воркеры работают на разных машинах.
• Приоритеты и отложенная доставка — стандартный канал — FIFO, приоритетов нет.
• Повторная обработка (retry с backoff) — канал не умеет «вернуть» сообщение.

Для retry внутри процесса проще всего обернуть вызов в цикл с экспоненциальной паузой, а не пытаться заново положить задачу в канал. Для персистентности и распределённости — NATS JetStream, RabbitMQ или Redis Streams, но это уже другая архитектура.

Worker pool^* — паттерн, при котором фиксированное число горутин (воркеров) забирает задачи из общей очереди (канала). Ограничивает параллелизм и делает нагрузку предсказуемой.

Что делать дальше

• Добавьте метрики: считайте обработанные задачи и время обработки через expvar или Prometheus-клиент. Без цифр вы не узнаете, хватает ли воркеров.
• Настройте размер пула по нагрузке: начните с числа ядер (runtime.NumCPU()) для CPU-bound задач и с числа допустимых соединений к базе для IO-bound.
• Добавьте таймаут на отдельную задачу через context.WithTimeout — зависшая задача не должна блокировать воркера навсегда.
• Прочитайте раздел Concurrency в Effective Go — там описана философия, стоящая за каналами.