Padrões de Concorrência em Go: 7 Padrões de Produção do Worker Pool ao Pipeline

编程语言

Quando Vazamentos de Goroutine Encontram Concorrência Sem Limites: Um Pesadelo em Produção

3 da manhã, alerta de OOM em produção. A investigação revela: um handler HTTP lançando goroutines sem limites — 3 goroutines por requisição, QPS 1000 significa 3.000, e em 10 minutos, 180.000 goroutines vazadas consomem toda a memória. Pior ainda, essas goroutines mantêm referências de canais que bloqueiam o GC, eventualmente crashando o nó inteiro.

Este não é um caso isolado. As primitivas de concorrência do Go são simples — go func() lança concorrência em uma linha — mas programação de concorrência em produção vai muito além de apenas iniciar goroutines. Você precisa controlar níveis de concorrência, lidar com propagação de erros, implementar desligamento graceful e prevenir vazamentos de recursos. Este artigo cobre 7 padrões de concorrência de nível produtivo para ajudá-lo a construir serviços concorrentes robustos em Go.

Referência de Conceitos Centrais

Primitiva Propósito Características Principais Caso de Uso Típico
goroutine Unidade de execução concorrente leve Stack crescente (2KB inicial), agendamento pelo runtime do Go Qualquer tarefa que requer execução concorrente
channel Comunicação entre goroutines Type-safe, com/sem buffer, fechável Passagem de dados, notificação de sinais, coleta de resultados
sync.WaitGroup Esperar por um grupo de goroutines Tríade Add/Done/Wait Espera de tarefas em lote, fan-out concorrente
sync.Mutex Exclusão mútua para estado compartilhado Usável com valor zero, suporte a TryLock (Go 1.18+) Contadores, atualizações de cache, hot-reload de config
context.Context Propagar cancelamento e timeout Imutável, só pode derivar contextos filhos Timeout de requisição, desligamento graceful, tracing
errgroup.Group Execução concorrente + coleta de erros Primeiro erro cancela todas as goroutines Chamadas de API em lote, busca paralela de dados
semaphore.Weighted Limitação de taxa com semáforo ponderado Suporta pesos, aquisição com timeout Limitação de API, controle de cota de recursos

5 Desafios da Programação de Concorrência em Produção

Desafio 1: Concorrência Sem Limites Causa Esgotamento de Recursos

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

Cada URL cria uma goroutine — 10.000 URLs significa 10.000 conexões concorrentes. Pools de conexões com banco de dados são esgotados, serviços downstream são sobrecarregados, a memória dispara.

Desafio 2: Vazamentos de Goroutine

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

Se o channel nunca for fechado, esta goroutine nunca sai. Em serviços de longa duração, goroutines vazadas se acumulam continuamente.

Desafio 3: Erros Silenciosamente Ignorados

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()
}

Erros são apenas logados — o chamador não tem ideia das falhas. Se 2 de 3 URLs falharem, o chamador assume que todas tiveram sucesso.

Desafio 4: Sem Desligamento Graceful

Quando o serviço recebe SIGTERM, as goroutines em execução são interrompidas à força. Escritas em andamento podem corromper dados, transações em progresso podem ficar incompletas.

Desafio 5: Dificuldade em Compor Padrões de Concorrência

Worker Pool precisa de limitação de taxa, Pipeline precisa de propagação de erros, Fan-out precisa de agregação de resultados. Implementar cada padrão isoladamente não é difícil, mas combiná-los em um serviço cria interações que facilmente levam a deadlocks.

7 Padrões de Concorrência de Nível Produtivo

Padrão 1: Worker Pool — Pool de Goroutines Limitado

Worker Pool é o padrão de concorrência mais fundamental e importante. Ideia central: um número fixo de workers retira tarefas de uma fila de tarefas, evitando concorrência sem limites.

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()
}

Exemplo de uso:

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()
}

Pontos-chave de design:

  • Contagem fixa de workers previne explosão de goroutines
  • Channel de tarefas com buffer serve como fila de tarefas
  • Suporte a Context para cancelamento, workers podem sair gracefulmente
  • Submit é non-blocking, retorna false quando a fila está cheia

Padrão 2: Fan-out/Fan-in — Dispersão-Coleta Paralela

Fan-out distribui uma fonte de dados para múltiplas goroutines para processamento paralelo, Fan-in mescla resultados de múltiplas goroutines em um único channel.

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
}

Exemplo de uso — processamento paralelo de pedidos:

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

Pontos-chave de design:

  • Suporte a genéricos (Go 1.18+), aplicável a qualquer tipo
  • Cada goroutine de fan-out consome independentemente do channel fonte
  • Fan-in usa WaitGroup para esperar que todos os channels de entrada fechem
  • Todas as goroutines podem sair no cancelamento do contexto

Padrão 3: Pipeline — Processamento em Estágios

Pipeline decompõe processamento complexo em estágios, cada um executando como uma goroutine conectada por channels.

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
    }
}

Exemplo de uso — pipeline de processamento de dados:

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)
        }
    }()
}

Pontos-chave de design:

  • Cada estágio é uma goroutine independente, pode escalar independentemente
  • Channels fornecem backpressure entre estágios
  • Erros tratados dentro dos estágios, não interrompem o pipeline inteiro
  • Todos os estágios saem gracefulmente no cancelamento do contexto

Padrão 4: errgroup — Tratamento de Erros Concorrentes

errgroup é a versão com consciência de erros do sync.WaitGroup: o primeiro erro cancela todas as goroutines.

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 com limite de concorrência:

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
}

Pontos-chave de design:

  • errgroup.WithContext propaga automaticamente o cancelamento
  • g.SetLimit(n) controla a concorrência máxima (Go 1.20+)
  • O primeiro erro cancela todas as goroutines em andamento
  • Captura de variável em closure requer i, url := i, url

Padrão 5: Semáforo — Limitação de Taxa

semaphore.Weighted fornece semáforos ponderados para alocação de recursos com diferentes pesos.

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()
}

Exemplo de uso — limitação de taxa de API em camadas:

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()
    })
}

Semáforo com timeout:

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()
}

Pontos-chave de design:

  • Semáforo ponderado, operações diferentes consomem cotas diferentes
  • Acquire suporta cancelamento de contexto e timeout
  • Release deve ser pareado com Acquire
  • Adequado para limitação de API em camadas, controle de cota de recursos

Padrão 6: Cancelamento e Timeout de Contexto

Context é a "linha da vida" da programação de concorrência em Go, usado para propagar sinais de cancelamento, timeouts e deadlines.

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
}

Padrão de desligamento graceful:

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()
}

Pontos-chave de design:

  • WithTimeout/WithDeadline para definir timeouts
  • WithCancel para controle manual de cancelamento
  • signal.NotifyContext para escutar sinais do sistema
  • O cancelamento do contexto propaga para todas as goroutines filhas
  • defer cancel() previne vazamentos de contexto

Padrão 7: Serviço Concorrente em Produção — Combinando Padrões

Combine os padrões acima para construir um serviço concorrente de nível produtivo: 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
}

Pontos-chave de design:

  • errgroup controla concorrência + propagação de erros
  • Channels para distribuição de tarefas e coleta de resultados
  • Context para timeout e cancelamento unificados
  • WaitGroup garante que todos os workers completem antes de fechar o channel de resultados
  • Design em camadas: agendamento (errgroup) + execução (worker goroutines) + coleta (range resultCh)

5 Armadilhas Comuns e Correções

Armadilha 1: Captura de Variável em Closure

❌ Errado:

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

Todas as goroutines compartilham a mesma variável url, buscando no final a última URL.

✅ Correto:

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

Armadilha 2: Channel Não Fechado Causa Vazamento de Goroutine

❌ Errado:

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

Consumidores iterando sobre o channel nunca terminam, goroutine vaza.

✅ Correto:

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

Armadilha 3: Posição Incorreta do WaitGroup Add

❌ Errado:

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

As goroutines podem ainda não ter alcançado wg.Add(1), e o wg.Wait() da goroutine principal passa imediatamente.

✅ Correto:

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

Armadilha 4: Vazamento do time.After no Select

❌ Errado:

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

Cada select cria um novo channel time.After. Chamadas frequentes vazam timers.

✅ Correto:

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

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

Armadilha 5: Cópia de Mutex

❌ Errado:

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

func copyMap(m SafeMap) SafeMap {
    return m
}

O valor zero do Mutex é "desbloqueado". Copiar um Mutex bloqueado causa deadlock ou data race.

✅ Correto:

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

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

Referência de Diagnóstico de Erros

Sintoma Possível Causa Investigação Solução
fatal error: all goroutines are asleep - deadlock! Todas as goroutines bloqueadas, sem goroutines ativas Verifique pareamento de leitura/escrita em channels, select default Garanta que channels tenham produtores e consumidores, adicione cancelamento de contexto
Contagem de goroutines cresce continuamente Vazamento de goroutines, channels não fechados Monitoramento com runtime.NumGoroutine(), pprof goroutine profile Garanta que todas as goroutines tenham caminhos de saída, defer close(ch)
Memória cresce continuamente Goroutines mantêm referências a objetos grandes, acúmulo em channels pprof heap profile, verifique tamanho do buffer do channel Limite o buffer do channel, libere referências prontamente
Erros frequentes de context canceled Contexto upstream cancelado, timeout muito curto Verifique a cadeia de contextos, confirme se o timeout é razoável Ajuste o timeout, distinga erros de negócio de erros de timeout
Resultados concorrentes faltando Goroutine sai antes de escrever no channel Verifique lógica de saída da goroutine, confirme fechamento do channel Use WaitGroup para garantir que todas as goroutines completem antes de fechar o channel
race condition detectada Múltiplas goroutines leem/escrevem variáveis compartilhadas go test -race, verifique variáveis globais e capturas em closures Use Mutex/RWMutex para proteção, ou mude para comunicação via channel
Channel bloqueado, sem resposta Descompasso de velocidade entre produtor/consumidor Verifique tamanho do buffer do channel, monitore taxas de produção/consumo Aumente o buffer, use select+default para envio non-blocking
Falha no desligamento graceful Goroutines não respondem ao cancelamento de contexto Verifique se as goroutines fazem select em ctx.Done() Garanta que todas as goroutines de longa duração verifiquem ctx.Done()
errgroup retorna apenas um erro errgroup foi projetado para cancelamento no primeiro erro Verifique se você precisa coletar todos os erros Use um coletor de erros personalizado ou múltiplos errgroups
Timeout na aquisição do semáforo Limite de concorrência muito restrito, requisições enfileiradas por muito tempo Monitore tempo de espera do semáforo, ajuste cotas de peso Aumente a capacidade do semáforo, otimize a alocação de pesos

Técnicas Avançadas de Otimização

Otimização 1: Worker Pool Dinâmico

Ajuste dinamicamente a contagem de workers com base na carga do sistema:

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()
}

Otimização 2: Batching

Combine múltiplas tarefas pequenas em uma operação em lote para reduzir overhead de syscall e 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
}

Otimização 3: Passagem por Channel Zero-Copy

Use ponteiros e sync.Pool para reduzir alocação de memória durante passagem por channels:

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
}

Comparação de Padrões de Concorrência

Característica goroutine+channel sync.WaitGroup errgroup Worker Pool
Controle de concorrência Sem controle embutido Sem controle embutido SetLimit Contagem fixa de workers
Tratamento de erros Implementação manual Implementação manual Cancelamento automático no primeiro erro Implementação manual
Backpressure Buffer do channel Nenhum Nenhum Channel de tarefas
Propagação de cancelamento Context manual Context manual Context automático Context manual
Coleta de resultados Recebimento por channel Variáveis compartilhadas Retorna slice Recebimento por channel
Caso de uso Processamento de dados em stream Espera de tarefas em lote Chamadas de API concorrentes Processamento de tarefas com limitação de taxa
Complexidade Média Baixa Baixa Média
Contagem de goroutines Variável Variável Limitada Fixa
Recomendação para produção ⭐⭐⭐ ⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐

Resumo

Programação de concorrência em Go não é sobre "apenas fazer rodar" — é sobre responder quatro perguntas: Quantas goroutines estão em execução? Quando elas saem? O que acontece em caso de erro? Como os recursos são recuperados? Worker Pool responde "quantas", Context responde "quando sair", errgroup responde "o que em caso de erro", e defer close() responde "como recuperar". Domine estes 7 padrões, e você terá a metodologia central para programação de concorrência em Go de nível produtivo.

Ferramentas Recomendadas

  • Formatador JSON — Formate respostas JSON de serviços concorrentes, depure rapidamente problemas de estrutura de dados
  • Calculadora de Hash — Compute assinaturas de requisição e checksums de dados, garanta consistência de dados em requisições concorrentes
  • Codificação/Decodificação Base64 — Trate codificação de dados binários em serviços concorrentes

Experimente estas ferramentas executadas localmente no navegador — nenhum cadastro necessário →

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