Паттерны конкурентности Go: 7 продакшен-паттернов от Worker Pool до Pipeline

编程语言

Когда утечки горутин встречаются с неограниченной конкурентностью: продакшен-кошмар

3 часа ночи, продакшен-алерт OOM. Расследование показывает: HTTP-обработчик запускает горутины без ограничений — 3 горутины на запрос, QPS 1000 означает 3 000, а за 10 минут 180 000 утечек горутин потребляют всю память. Хуже того, эти горутины удерживают ссылки на каналы, блокирующие GC, что в итоге приводит к краху всего узла.

Это не единичный случай. Примитивы конкурентности Go просты — go func() запускает конкурентность в одну строку — но продакшен-конкурентное программирование — это гораздо больше, чем просто запуск горутин. Вам нужно контролировать уровень конкурентности, обрабатывать распространение ошибок, реализовывать грациозное завершение и предотвращать утечки ресурсов. В этой статье рассматриваются 7 продакшен-паттернов конкурентности, которые помогут вам создавать надёжные конкурентные сервисы на Go.

Справочник базовых концепций

Примитив Назначение Ключевые особенности Типичный случай использования
goroutine Лёгкая единица конкурентного выполнения Расширяемый стек (2KB начальный), планирование Go runtime Любая задача, требующая конкурентного выполнения
channel Межгорутинное взаимодействие Типобезопасные, буферизованные/небуферизованные, закрываемые Передача данных, сигнальные уведомления, сбор результатов
sync.WaitGroup Ожидание группы горутин Трио Add/Done/Wait Ожидание пакетных задач, конкурентный fan-out
sync.Mutex Взаимное исключение для общего состояния Годен с нулевым значением, поддержка TryLock (Go 1.18+) Счётчики, обновления кэша, горячая перезагрузка конфигурации
context.Context Распространение отмены и таймаута Неизменяемый, можно только порождать дочерние контексты Таймаут запроса, грациозное завершение, трассировка
errgroup.Group Конкурентное выполнение + сбор ошибок Первая ошибка отменяет все горутины Пакетные вызовы API, параллельная загрузка данных
semaphore.Weighted Взвешенная семафорная_rate-лимитинг Поддержка весов, получение с учётом таймаута Rate-лимитинг API, контроль квот ресурсов

5 проблем продакшен-конкурентного программирования

Проблема 1: Неограниченная конкурентность вызывает исчерпание ресурсов

func handleRequests(urls []string) {
    for _, url := range urls {
        go fetch(url)
    }
}

Каждый URL порождает горутину — 10 000 URL означает 10 000 конкурентных соединений. Пулы соединений с БД исчерпываются, нижестоящие сервисы раздавливаются, память взлетает.

Проблема 2: Утечки горутин

func process(ch <-chan int) {
    for {
        val := <-ch
        fmt.Println(val)
    }
}

Если канал никогда не закрывается, эта горутина никогда не завершается. В долго работающих сервисах утечки горутин накапливаются непрерывно.

Проблема 3: Ошибки молча проглатываются

func fetchAll(urls []string) {
    var wg sync.WaitGroup
    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            resp, err := http.Get(u)
            if err != nil {
                log.Printf("fetch %s failed: %v", u, err)
                return
            }
            process(resp)
        }(url)
    }
    wg.Wait()
}

Ошибки только логируются — вызывающая сторона не знает о неудачах. Если 2 из 3 URL завершаются ошибкой, вызывающая сторона предполагает, что все успешны.

Проблема 4: Отсутствие грациозного завершения

Когда сервис получает SIGTERM, запущенные горутины принудительно прерываются. Текущие записи могут повредить данные, выполняемые транзакции могут остаться незавершёнными.

Проблема 5: Сложность комбинирования паттернов конкурентности

Worker Pool нуждается в rate-лимитинге, Pipeline нуждается в распространении ошибок, Fan-out нуждается в агрегации результатов. Реализация каждого паттерна по отдельности не сложна, но их комбинирование в одном сервисе создаёт взаимодействия, которые легко приводят к дедлокам.

7 продакшен-паттернов конкурентности

Паттерн 1: Worker Pool — ограниченный пул горутин

Worker Pool — самый фундаментальный и важный паттерн конкурентности. Основная идея: фиксированное количество воркеров забирает задачи из очереди задач, избегая неограниченной конкурентности.

package workerpool

import (
    "context"
    "sync"
)

type Task func(ctx context.Context) error

type Pool struct {
    workers int
    tasks   chan Task
    wg      sync.WaitGroup
}

func NewPool(workers int, bufferSize int) *Pool {
    return &Pool{
        workers: workers,
        tasks:   make(chan Task, bufferSize),
    }
}

func (p *Pool) Start(ctx context.Context) {
    for i := 0; i < p.workers; i++ {
        p.wg.Add(1)
        go func(workerID int) {
            defer p.wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case task, ok := <-p.tasks:
                    if !ok {
                        return
                    }
                    _ = task(ctx)
                }
            }
        }(i)
    }
}

func (p *Pool) Submit(task Task) bool {
    select {
    case p.tasks <- task:
        return true
    default:
        return false
    }
}

func (p *Pool) Stop() {
    close(p.tasks)
    p.wg.Wait()
}

Пример использования:

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    pool := NewPool(10, 100)
    pool.Start(ctx)

    urls := []string{
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    }

    for _, url := range urls {
        u := url
        pool.Submit(func(ctx context.Context) error {
            return fetchURL(ctx, u)
        })
    }

    pool.Stop()
}

Ключевые моменты проектирования:

  • Фиксированное количество воркеров предотвращает взрыв горутин
  • Буферизованный канал задач служит очередью задач
  • Поддержка Context для отмены, воркеры могут завершаться грациозно
  • Submit неблокирующий, возвращает false при полной очереди

Паттерн 2: Fan-out/Fan-in — параллельный scatter-gather

Fan-out распределяет источник данных по нескольким горутинам для параллельной обработки, Fan-in объединяет результаты из нескольких горутин в один канал.

package fan

import (
    "context"
    "sync"
)

func FanOut[T any](ctx context.Context, source <-chan T, workers int) []<-chan T {
    channels := make([]<-chan T, workers)
    for i := 0; i < workers; i++ {
        ch := make(chan T)
        channels[i] = ch
        go func() {
            defer close(ch)
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-source:
                    if !ok {
                        return
                    }
                    select {
                    case ch <- val:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }()
    }
    return channels
}

func FanIn[T any](ctx context.Context, channels ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    for _, ch := range channels {
        go func(c <-chan T) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-c:
                    if !ok {
                        return
                    }
                    select {
                    case out <- val:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

Пример использования — параллельная обработка заказов:

func processOrders(ctx context.Context, orders <-chan Order) <-chan Result {
    workers := FanOut(ctx, orders, 5)
    return FanIn(ctx, workers...)
}

Ключевые моменты проектирования:

  • Поддержка дженериков (Go 1.18+), применимо к любому типу
  • Каждая fan-out горутина независимо потребляет из исходного канала
  • Fan-in использует WaitGroup для ожидания закрытия всех входных каналов
  • Все горутины могут завершиться при отмене контекста

Паттерн 3: Pipeline — поэтапная обработка

Pipeline декомпозирует сложную обработку на этапы, каждый из которых выполняется как горутина, соединённая каналами.

package pipeline

import (
    "context"
)

type Stage[In any, Out any] func(ctx context.Context, in <-chan In) <-chan Out

func NewPipeline[In any, Out any](
    ctx context.Context,
    source <-chan In,
    stages ...Stage[In, In],
) <-chan Out {
    current := source
    for _, stage := range stages {
        current = stage(ctx, current)
    }
    return any(current).(<-chan Out)
}

func NewStage[In any, Out any](
    process func(ctx context.Context, in In) (Out, error),
    bufferSize int,
) Stage[In, Out] {
    return func(ctx context.Context, in <-chan In) <-chan Out {
        out := make(chan Out, bufferSize)
        go func() {
            defer close(out)
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-in:
                    if !ok {
                        return
                    }
                    result, err := process(ctx, val)
                    if err != nil {
                        continue
                    }
                    select {
                    case out <- result:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }()
        return out
    }
}

Пример использования — конвейер обработки данных:

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

    raw := make(chan RawData, 100)

    validate := NewStage[RawData, ValidData](func(ctx context.Context, in RawData) (ValidData, error) {
        if err := in.Validate(); err != nil {
            return ValidData{}, err
        }
        return in.ToValid(), nil
    }, 50)

    enrich := NewStage[ValidData, EnrichedData](func(ctx context.Context, in ValidData) (EnrichedData, error) {
        return fetchExtraInfo(ctx, in)
    }, 50)

    transform := NewStage[EnrichedData, FinalData](func(ctx context.Context, in EnrichedData) (FinalData, error) {
        return in.Transform()
    }, 50)

    result := NewPipeline(ctx, raw, validate, enrich, transform)

    go func() {
        for r := range result {
            saveToDB(r)
        }
    }()
}

Ключевые моменты проектирования:

  • Каждый этап — независимая горутина, может масштабироваться независимо
  • Каналы обеспечивают обратное давление между этапами
  • Ошибки обрабатываются внутри этапов, не прерывают весь конвейер
  • Все этапы грациозно завершаются при отмене контекста

Паттерн 4: errgroup — конкурентная обработка ошибок

errgroup — это версия sync.WaitGroup с поддержкой ошибок: первая ошибка отменяет все горутины.

package main

import (
    "context"
    "fmt"
    "net/http"
    "time"

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

type FetchResult struct {
    URL  string
    Body string
    Size int
}

func fetchMultiple(ctx context.Context, urls []string) ([]FetchResult, error) {
    g, ctx := errgroup.WithContext(ctx)
    results := make([]FetchResult, len(urls))

    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
            if err != nil {
                return fmt.Errorf("create request %s: %w", url, err)
            }

            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return fmt.Errorf("fetch %s: %w", url, err)
            }
            defer resp.Body.Close()

            if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("fetch %s: status %d", url, resp.StatusCode)
            }

            body, err := io.ReadAll(resp.Body)
            if err != nil {
                return fmt.Errorf("read %s: %w", url, err)
            }

            results[i] = FetchResult{
                URL:  url,
                Body: string(body),
                Size: len(body),
            }
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

errgroup с ограничением конкурентности:

func fetchWithLimit(ctx context.Context, urls []string, maxConcurrent int) ([]FetchResult, error) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(maxConcurrent)

    results := make([]FetchResult, len(urls))
    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            results[i], _ = fetchOne(ctx, url)
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

Ключевые моменты проектирования:

  • errgroup.WithContext автоматически распространяет отмену
  • g.SetLimit(n) контролирует максимальную конкурентность (Go 1.20+)
  • Первая ошибка отменяет все выполняемые горутины
  • Захват переменных замыкания требует i, url := i, url

Паттерн 5: Semaphore — rate-лимитинг

semaphore.Weighted предоставляет взвешенные семафоры для распределения ресурсов с разными весами.

package ratelimit

import (
    "context"
    "fmt"

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

type RateLimiter struct {
    sem *semaphore.Weighted
}

func NewRateLimiter(maxWeight int64) *RateLimiter {
    return &RateLimiter{
        sem: semaphore.NewWeighted(maxWeight),
    }
}

func (r *RateLimiter) Do(ctx context.Context, weight int64, fn func() error) error {
    if err := r.sem.Acquire(ctx, weight); err != nil {
        return fmt.Errorf("acquire semaphore: %w", err)
    }
    defer r.sem.Release(weight)
    return fn()
}

Пример использования — многоуровневый rate-лимитинг API:

func main() {
    limiter := NewRateLimiter(100)

    err := limiter.Do(context.Background(), 10, func() error {
        return callLightAPI()
    })

    err = limiter.Do(context.Background(), 50, func() error {
        return callHeavyAPI()
    })
}

Семафор с таймаутом:

func (r *RateLimiter) DoWithTimeout(ctx context.Context, weight int64, timeout time.Duration, fn func() error) error {
    acquireCtx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    if err := r.sem.Acquire(acquireCtx, weight); err != nil {
        return fmt.Errorf("semaphore acquire timeout: %w", err)
    }
    defer r.sem.Release(weight)
    return fn()
}

Ключевые моменты проектирования:

  • Взвешенный семафор, разные операции потребляют разные квоты
  • Acquire поддерживает отмену контекста и таймаут
  • Release должен быть парным с Acquire
  • Подходит для многоуровневого rate-лимитинга API, контроля квот ресурсов

Паттерн 6: Отмена и таймаут контекста

Context — это «линия жизни» конкурентного программирования на Go, используемая для распространения сигналов отмены, таймаутов и дедлайнов.

package ctxutil

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

type Result struct {
    Data  string
    Error error
}

func FetchWithTimeout(ctx context.Context, url string, timeout time.Duration) (string, error) {
    ctx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return "", fmt.Errorf("create request: %w", err)
    }

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        if ctx.Err() == context.DeadlineExceeded {
            return "", fmt.Errorf("fetch %s timed out after %v: %w", url, timeout, ctx.Err())
        }
        return "", fmt.Errorf("fetch %s: %w", url, err)
    }
    defer resp.Body.Close()

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return "", fmt.Errorf("read response: %w", err)
    }
    return string(body), nil
}

func BatchFetch(ctx context.Context, urls []string, timeout time.Duration) []Result {
    results := make([]Result, len(urls))
    var wg sync.WaitGroup

    for i, url := range urls {
        wg.Add(1)
        go func(idx int, u string) {
            defer wg.Done()
            data, err := FetchWithTimeout(ctx, u, timeout)
            results[idx] = Result{Data: data, Error: err}
        }(i, url)
    }

    wg.Wait()
    return results
}

Паттерн грациозного завершения:

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

    server := &http.Server{Addr: ":8080"}

    go func() {
        <-ctx.Done()
        shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()
        _ = server.Shutdown(shutdownCtx)
    }()

    _ = server.ListenAndServe()
}

Ключевые моменты проектирования:

  • WithTimeout/WithDeadline для установки таймаутов
  • WithCancel для ручного управления отменой
  • signal.NotifyContext для прослушивания системных сигналов
  • Отмена контекста распространяется на все дочерние горутины
  • defer cancel() предотвращает утечки контекста

Паттерн 7: Продакшен-конкурентный сервис — комбинирование паттернов

Объедините вышеописанные паттерны для создания продакшен-готового конкурентного сервиса: Worker Pool + Pipeline + errgroup + Context.

package concurrencyservice

import (
    "context"
    "fmt"
    "sync"
    "time"

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

type Service struct {
    workers    int
    bufferSize int
    timeout    time.Duration
}

func NewService(workers, bufferSize int, timeout time.Duration) *Service {
    return &Service{
        workers:    workers,
        bufferSize: bufferSize,
        timeout:    timeout,
    }
}

type Job struct {
    ID    string
    Input any
}

type Output struct {
    Job    Job
    Result any
    Err    error
}

func (s *Service) Process(ctx context.Context, jobs []Job) ([]Output, error) {
    ctx, cancel := context.WithTimeout(ctx, s.timeout)
    defer cancel()

    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(s.workers)

    jobCh := make(chan Job, s.bufferSize)
    resultCh := make(chan Output, s.bufferSize)

    g.Go(func() error {
        defer close(jobCh)
        for _, job := range jobs {
            select {
            case <-ctx.Done():
                return ctx.Err()
            case jobCh <- job:
            }
        }
        return nil
    })

    var processWg sync.WaitGroup
    for i := 0; i < s.workers; i++ {
        processWg.Add(1)
        go func() {
            defer processWg.Done()
            for job := range jobCh {
                output := Output{Job: job}
                output.Result, output.Err = s.processOne(ctx, job)
                select {
                case resultCh <- output:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }

    go func() {
        processWg.Wait()
        close(resultCh)
    }()

    var outputs []Output
    for result := range resultCh {
        outputs = append(outputs, result)
    }

    if err := g.Wait(); err != nil {
        return outputs, fmt.Errorf("service process: %w", err)
    }
    return outputs, nil
}

func (s *Service) processOne(ctx context.Context, job Job) (any, error) {
    select {
    case <-ctx.Done():
        return nil, ctx.Err()
    default:
    }
    return fmt.Sprintf("processed-%s", job.ID), nil
}

Ключевые моменты проектирования:

  • errgroup контролирует конкурентность + распространение ошибок
  • Каналы для распределения задач и сбора результатов
  • Context для единого таймаута и отмены
  • WaitGroup гарантирует завершение всех воркеров перед закрытием канала результатов
  • Многоуровневое проектирование: планирование (errgroup) + выполнение (воркер-горутины) + сбор (range resultCh)

5 распространённых ловушек и их исправления

Ловушка 1: Захват переменных замыкания

❌ Неправильно:

for _, url := range urls {
    go func() {
        fetch(url)
    }()
}

Все горутины разделяют одну и ту же переменную url, в итоге запрашивая последний URL.

✅ Правильно:

for _, url := range urls {
    url := url
    go func() {
        fetch(url)
    }()
}

Ловушка 2: Незакрытый канал вызывает утечку горутин

❌ Неправильно:

func producer() <-chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 100; i++ {
            ch <- i
        }
    }()
    return ch
}

Потребители, итерирующие по каналу через range, никогда не завершатся, горутина утекает.

✅ Правильно:

func producer() <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        for i := 0; i < 100; i++ {
            ch <- i
        }
    }()
    return ch
}

Ловушка 3: Ошибка позиции WaitGroup Add

❌ Неправильно:

var wg sync.WaitGroup
for i := 0; i < 10; i++ {
    go func() {
        wg.Add(1)
        defer wg.Done()
        doWork()
    }()
}
wg.Wait()

Горутины могли ещё не достичь wg.Add(1), и wg.Wait() основной горутины проходит немедленно.

✅ Правильно:

var wg sync.WaitGroup
for i := 0; i < 10; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        doWork()
    }()
}
wg.Wait()

Ловушка 4: Утечка time.After в Select

❌ Неправильно:

select {
case result := <-ch:
    process(result)
case <-time.After(5 * time.Second):
    return errors.New("timeout")
}

Каждый select создаёт новый канал time.After. Частые вызовы утекают таймеры.

✅ Правильно:

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

select {
case result := <-ch:
    process(result)
case <-ctx.Done():
    return ctx.Err()
}

Ловушка 5: Копирование Mutex

❌ Неправильно:

type SafeMap struct {
    mu   sync.Mutex
    data map[string]string
}

func copyMap(m SafeMap) SafeMap {
    return m
}

Нулевое значение Mutex — «разблокировано». Копирование заблокированного Mutex вызывает дедлок или гонку данных.

✅ Правильно:

type SafeMap struct {
    mu   *sync.Mutex
    data map[string]string
}

func NewSafeMap() *SafeMap {
    return &SafeMap{
        mu:   &sync.Mutex{},
        data: make(map[string]string),
    }
}

Справочник по устранению ошибок

Симптом Возможная причина Расследование Решение
fatal error: all goroutines are asleep - deadlock! Все горутины заблокированы, нет активных горутин Проверьте парность чтения/записи каналов, select default Убедитесь, что каналы имеют производителей и потребителей, добавьте отмену контекста
Количество горутин постоянно растёт Утечка горутин, незакрытые каналы Мониторинг runtime.NumGoroutine(), pprof goroutine profile Убедитесь, что все горутины имеют пути выхода, defer close(ch)
Память постоянно растёт Горутины удерживают ссылки на большие объекты, накопление в каналах pprof heap profile, проверьте размер буфера каналов Ограничьте буфер каналов, своевременно освобождайте ссылки
Частые ошибки context canceled Вышестоящий контекст отменён, таймаут слишком короткий Проверьте цепочку контекстов, убедитесь, что таймаут разумный Скорректируйте таймаут, различайте бизнес-ошибки и ошибки таймаута
Пропадают конкурентные результаты Горутина завершается до записи в канал Проверьте логику выхода горутин, подтвердите закрытие канала Используйте WaitGroup для гарантии завершения всех горутин перед закрытием канала
Обнаружена race condition Несколько горутин читают/пишут общие переменные go test -race, проверьте глобальные переменные и захваты замыканий Используйте Mutex/RWMutex для защиты или переключитесь на коммуникацию через каналы
Канал заблокирован, нет ответа Несоответствие скорости производителя/потребителя Проверьте размер буфера канала, отслеживайте скорости производства/потребления Увеличьте буфер, используйте select+default для неблокирующей отправки
Сбой грациозного завершения Горутины не реагируют на отмену контекста Проверьте, выбирают ли горутины ctx.Done() в select Убедитесь, что все долго работающие горутины проверяют ctx.Done()
errgroup возвращает только одну ошибку errgroup разработан для отмены по первой ошибке Проверьте, нужно ли собирать все ошибки Используйте пользовательский сборщик ошибок или несколько errgroup
Таймаут получения семафора Ограничение конкурентности слишком строгое, запросы стоят в очереди слишком долго Отслеживайте время ожидания семафора, скорректируйте весовые квоты Увеличьте ёмкость семафора, оптимизируйте распределение весов

Продвинутые техники оптимизации

Оптимизация 1: Динамический Worker Pool

Динамическая регулировка количества воркеров на основе нагрузки системы:

package dynamicpool

import (
    "context"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)

type DynamicPool struct {
    minWorkers int64
    maxWorkers int64
    active     atomic.Int64
    tasks      chan Task
    wg         sync.WaitGroup
    adjustTick *time.Ticker
}

func NewDynamicPool(minW, maxW int) *DynamicPool {
    return &DynamicPool{
        minWorkers: int64(minW),
        maxWorkers: int64(maxW),
        tasks:      make(chan Task, maxW*2),
        adjustTick: time.NewTicker(5 * time.Second),
    }
}

func (p *DynamicPool) Start(ctx context.Context) {
    for i := 0; i < int(p.minWorkers); i++ {
        p.addWorker(ctx)
    }

    go func() {
        for {
            select {
            case <-ctx.Done():
                p.adjustTick.Stop()
                return
            case <-p.adjustTick.C:
                p.adjustWorkers(ctx)
            }
        }
    }()
}

func (p *DynamicPool) addWorker(ctx context.Context) {
    p.active.Add(1)
    p.wg.Add(1)
    go func() {
        defer func() {
            p.active.Add(-1)
            p.wg.Done()
        }()
        for task := range p.tasks {
            _ = task(ctx)
        }
    }()
}

func (p *DynamicPool) adjustWorkers(ctx context.Context) {
    current := p.active.Load()
    queueLen := len(p.tasks)

    if queueLen > int(current) && current < p.maxWorkers {
        p.addWorker(ctx)
    } else if queueLen == 0 && current > p.minWorkers {
        for i := 0; i < 5 && int(p.active.Load()) > int(p.minWorkers); i++ {
            p.tasks <- nil
        }
    }
}

func (p *DynamicPool) Stop() {
    close(p.tasks)
    p.wg.Wait()
}

Оптимизация 2: Батчинг

Объединение нескольких мелких задач в одну пакетную операцию для снижения накладных расходов на syscall и IO:

package batcher

import (
    "context"
    "sync"
    "time"
)

type Batcher[T any, R any] struct {
    batchSize     int
    flushInterval time.Duration
    handler       func(ctx context.Context, batch []T) ([]R, error)
    mu            sync.Mutex
    buffer        []T
    results       []R
}

func NewBatcher[T any, R any](
    batchSize int,
    flushInterval time.Duration,
    handler func(ctx context.Context, batch []T) ([]R, error),
) *Batcher[T, R] {
    return &Batcher[T, R]{
        batchSize:     batchSize,
        flushInterval: flushInterval,
        handler:       handler,
        buffer:        make([]T, 0, batchSize),
    }
}

func (b *Batcher[T, R]) Add(ctx context.Context, item T) (R, error) {
    b.mu.Lock()
    b.buffer = append(b.buffer, item)

    if len(b.buffer) >= b.batchSize {
        batch := b.buffer
        b.buffer = make([]T, 0, b.batchSize)
        b.mu.Unlock()

        results, err := b.handler(ctx, batch)
        if err != nil {
            var zero R
            return zero, err
        }
        b.results = append(b.results, results...)
        var zero R
        if len(results) > 0 {
            return results[0], nil
        }
        return zero, nil
    }
    b.mu.Unlock()

    var zero R
    return zero, nil
}

func (b *Batcher[T, R]) Flush(ctx context.Context) ([]R, error) {
    b.mu.Lock()
    batch := b.buffer
    b.buffer = make([]T, 0, b.batchSize)
    b.mu.Unlock()

    if len(batch) == 0 {
        return b.results, nil
    }

    results, err := b.handler(ctx, batch)
    if err != nil {
        return nil, err
    }
    b.results = append(b.results, results...)
    return b.results, nil
}

Оптимизация 3: Передача через канал без копирования

Использование указателей и sync.Pool для снижения выделения памяти при передаче через каналы:

package zerocopy

import (
    "sync"
)

type Buffer struct {
    Data []byte
}

var bufferPool = sync.Pool{
    New: func() any {
        return &Buffer{Data: make([]byte, 0, 4096)}
    },
}

func GetBuffer() *Buffer {
    return bufferPool.Get().(*Buffer)
}

func PutBuffer(buf *Buffer) {
    buf.Data = buf.Data[:0]
    bufferPool.Put(buf)
}

func ProcessPipeline(ctx context.Context, input <-chan []byte) <-chan *Buffer {
    out := make(chan *Buffer, 64)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                return
            case data, ok := <-input:
                if !ok {
                    return
                }
                buf := GetBuffer()
                buf.Data = append(buf.Data[:0], data...)

                select {
                case out <- buf:
                case <-ctx.Done():
                    PutBuffer(buf)
                    return
                }
            }
        }
    }()
    return out
}

Сравнение паттернов конкурентности

Свойство goroutine+channel sync.WaitGroup errgroup Worker Pool
Контроль конкурентности Нет встроенного контроля Нет встроенного контроля SetLimit Фиксированное количество воркеров
Обработка ошибок Ручная реализация Ручная реализация Автоотмена по первой ошибке Ручная реализация
Обратное давление Буфер канала Нет Нет Канал задач
Распространение отмены Ручной контекст Ручной контекст Авто контекст Ручной контекст
Сбор результатов Получение из канала Общие переменные Возврат среза Получение из канала
Случай использования Потоковая обработка данных Ожидание пакетных задач Конкурентные вызовы API Обработка задач с rate-лимитом
Сложность Средняя Низкая Низкая Средняя
Количество горутин Нефиксированное Нефиксированное Ограниченное Фиксированное
Продакшен-рекомендация ⭐⭐⭐ ⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐

Заключение

Конкурентное программирование на Go — это не просто «заставить это работать» — это ответ на четыре вопроса: Сколько горутин выполняется? Когда они завершаются? Что происходит при ошибке? Как освобождаются ресурсы? Worker Pool отвечает «сколько», Context отвечает «когда завершать», errgroup отвечает «что при ошибке», а defer close() отвечает «как освободить». Освойте эти 7 паттернов, и вы получите базовую методологию продакшен-конкурентного программирования на Go.

Рекомендуемые инструменты

  • JSON Formatter — Форматирование JSON-ответов конкурентных сервисов, быстрая отладка проблем со структурой данных
  • Hash Calculator — Вычисление подписей запросов и контрольных сумм данных, обеспечение согласованности данных при конкурентных запросах
  • Base64 Encode/Decode — Обработка кодирования бинарных данных в конкурентных сервисах

Попробуйте эти локальные браузерные инструменты — регистрация не требуется →

#Go#并发#goroutine#channel#sync#2026#并发模式