Продвинутая конкурентность в Go

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

Продвинутая конкурентность — это не про запуск большего количества горутин, а про их эффективное управление.

Паттерн Worker Pool

Представьте себе кассы в супермаркете. Вместо того чтобы открывать новую кассу для каждого покупателя (что привело бы к хаосу), магазин нанимает фиксированное число кассиров. Покупатели встают в общую очередь, а свободный кассир берет следующего. Это и есть пул воркеров (Worker Pool).

В Go воркеры — это горутины, которые читают задачи из одного канала и отправляют результаты в другой. Такой подход позволяет ограничить количество одновременно выполняемых задач, делая использование ресурсов предсказуемым.

package main

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

// worker — это наша "касса". Он читает задачи из канала jobs
// и отправляет результат в results.
func worker(id int, wg *sync.WaitGroup, jobs <-chan int, results chan<- int) {
	defer wg.Done()
	for j := range jobs {
		fmt.Printf("Воркер %d начал задачу %d\n", id, j)
		time.Sleep(time.Second) // Имитация работы
		fmt.Printf("Воркер %d закончил задачу %d\n", id, j)
		results <- j * 2
	}
}

func main() {
	const numJobs = 5
	const numWorkers = 3

	jobs := make(chan int, numJobs)
	results := make(chan int, numJobs)

	var wg sync.WaitGroup

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

	// Отправляем 5 задач в очередь
	for j := 1; j <= numJobs; j++ {
		jobs <- j
	}
	close(jobs) // Закрываем канал, чтобы воркеры завершились после выполнения всех задач

	wg.Wait() // Ждем, пока все воркеры закончат работу

	close(results)

	// Собираем результаты
	for a := range results {
		fmt.Printf("Получен результат: %d\n", a)
	}
}

Здесь мы создали 3 воркера для обработки 5 задач. Воркеры работают параллельно, но одновременно активны не более трех. Это предотвращает перегрузку системы. Такой паттерн незаменим при работе с ограниченными ресурсами, например, при выполнении HTTP-запросов или обращении к базе данных.

Конвейеры и Fan-out/Fan-in

Более сложные задачи можно разбить на этапы и организовать в конвейер (Pipeline). Каждая стадия конвейера — это горутина, которая принимает данные из одного канала, обрабатывает их и передает в следующий. Это позволяет строить гибкие и масштабируемые системы обработки данных.

Когда один из этапов конвейера становится узким местом, его можно распараллелить с помощью паттерна / Fan-in.

Fan-out (расширение): Одна горутина читает задачи из входного канала и распределяет их по нескольким воркерам.
Fan-in (сужение): Одна горутина собирает результаты работы всех воркеров в один выходной канал.

Такая архитектура позволяет динамически масштабировать производительность узких мест, просто добавляя больше воркеров на нужном этапе.

Управление жизненным циклом

Что произойдет, если пользователь отменит запрос, пока ваши горутины усердно работают? Без механизма отмены они продолжат выполняться, потребляя ресурсы впустую. Это называется утечкой горутин. В долгоживущих сервисах такие утечки могут привести к исчерпанию памяти и отказу системы.

Для решения этой проблемы в Go существует пакет context. позволяет передавать сигналы отмены, тайм-ауты и другие значения по цепочке вызовов, в том числе и в горутины.

Ключевое правило: контекст всегда передается как первый аргумент функции. Никогда не храните его в структурах.

package main

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

func longOperation(ctx context.Context, results chan<- string) {
	select {
	case <-time.After(5 * time.Second): // Имитация долгой работы
		results <- "Операция завершена успешно"
	case <-ctx.Done(): // Контекст был отменен
		results <- fmt.Sprintf("Операция отменена: %v", ctx.Err())
	}
}

func main() {
	// Создаем контекст с тайм-аутом в 2 секунды
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel() // Важно всегда вызывать cancel, чтобы освободить ресурсы

	results := make(chan string, 1)
	go longOperation(ctx, results)

	fmt.Println(<-results) // Выведет: Операция отменена: context deadline exceeded
}

В этом примере longOperation занимает 5 секунд, но мы создали контекст с тайм-аутом в 2 секунды. Благодаря select горутина немедленно прекратит работу, как только получит сигнал отмены через канал ctx.Done(). Это позволяет избежать ненужной работы и освободить ресурсы.

Обработка ошибок

Как собрать ошибки из множества параллельно работающих горутин? Если одна из них завершилась с ошибкой, нужно ли отменять остальные? Для элегантного решения этих вопросов существует пакет golang.org/x/sync/errgroup.

errgroup.Group позволяет запустить несколько горутин, дождаться их завершения и получить первую возникшую ошибку. Что еще важнее, при возникновении ошибки он автоматически отменяет контекст для всех остальных горутин в группе.

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"golang.org/x/sync/errgroup"
)

func main() {
	g, ctx := errgroup.WithContext(context.Background())

	urls := []string{
		"http://example.com",
		"http://invalid-url", // Эта задача завершится с ошибкой
		"http://example.org",
	}

	for _, url := range urls {
		url := url // Важно для замыкания в цикле
		g.Go(func() error {
			// Проверяем, не отменили ли нас еще до начала работы
			if ctx.Err() != nil {
				return ctx.Err()
			}

			if url == "http://invalid-url" {
				return errors.New("неверный URL")
			}

			fmt.Printf("Обработка %s\n", url)
			time.Sleep(2 * time.Second)
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		fmt.Printf("Произошла ошибка: %v\n", err)
	}
}

Как только горутина, обрабатывающая http://invalid-url, вернет ошибку, errgroup отменит контекст. Остальные горутины, которые еще могут работать, увидят это через ctx.Err() и смогут грациозно завершиться, не выполняя лишней работы. Метод g.Wait() дождется завершения всех горутин и вернет самую первую ошибку.

Освоение этих паттернов и инструментов позволит вам писать не просто конкурентный, а надежный, масштабируемый и эффективный код на Go, способный справляться с реальными нагрузками.

Quiz Questions 1/6

Какова основная проблема бесконтрольного создания горутин, например, с помощью go func() в цикле?

Quiz Questions 2/6

Какой паттерн concurrency лучше всего подходит для ограничения количества одновременных HTTP-запросов к внешнему API?

Golang для профи от теории к высоконагруженным системам

Golang для профи от теории к высоконагруженным системам

Продвинутая конкурентность в Go