Padrões de concorrência: fan-out, fan-in e pipelines

Os artigos anteriores (Select, timeouts e cancelamento de operações, sync.WaitGroup e sync.Mutex: coordenando goroutines, O pacote context: propagação de cancelamento e prazos) apresentaram as ferramentas: goroutines, channels, select, WaitGroup, Mutex e context. Este artigo mostra como combiná-las em padrões arquiteturais — estruturas reutilizáveis que resolvem classes inteiras de problemas concorrentes de forma elegante e previsível.

Padrões não são receitas rígidas. São vocabulário. Quando um engenheiro diz "vamos usar um pipeline com fan-out no estágio de processamento", toda a equipe entende imediatamente a estrutura, as garantias e as trocas envolvidas. Dominar esses padrões é comunicar intenção com precisão.

Pipeline revisitado: com cancelamento

O artigo Channels: comunicação entre goroutines introduziu pipelines. Aqui o padrão é refinado com cancelamento via context — essencial para pipelines em produção:

package main

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

// Estágio gerador — produz valores até o context ser cancelado
func gerar(ctx context.Context, valores ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, v := range valores {
            select {
            case out <- v:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Estágio transformador — aplica uma função a cada valor
func transformar(ctx context.Context, in <-chan int, fn func(int) int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for v := range in {
            select {
            case out <- fn(v):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Estágio filtro — passa apenas valores que satisfazem o predicado
func filtrar(ctx context.Context, in <-chan int, predicado func(int) bool) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for v := range in {
            if predicado(v) {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

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

    // Pipeline: gerar → dobrar → filtrar maiores que 10 → imprimir
    numeros := gerar(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    dobrados := transformar(ctx, numeros, func(n int) int { return n * 2 })
    maioresQue10 := filtrar(ctx, dobrados, func(n int) bool { return n > 10 })

    for v := range maioresQue10 {
        fmt.Println(v) // 12, 14, 16, 18, 20
    }
}

Cada estágio é uma goroutine independente. O context garante que, ao cancelar, todos os estágios encerram graciosamente sem goroutine leaks.

Fan-out: distribuindo trabalho

Fan-out distribui os itens de um único channel de entrada para múltiplos workers, processando em paralelo:

package main

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

type Tarefa struct {
    ID   int
    Dado string
}

type Resultado struct {
    TarefaID int
    Saida    string
    Duracao  time.Duration
}

func worker(
    ctx context.Context,
    id int,
    tarefas <-chan Tarefa,
    resultados chan<- Resultado,
    wg *sync.WaitGroup,
) {
    defer wg.Done()

    for {
        select {
        case tarefa, ok := <-tarefas:
            if !ok {
                return
            }
            inicio := time.Now()

            // Simula processamento com duração variável
            select {
            case <-time.After(time.Duration(50+tarefa.ID*10) * time.Millisecond):
            case <-ctx.Done():
                return
            }

            resultados <- Resultado{
                TarefaID: tarefa.ID,
                Saida:    fmt.Sprintf("worker%d processou: %s", id, tarefa.Dado),
                Duracao:  time.Since(inicio),
            }

        case <-ctx.Done():
            return
        }
    }
}

func fanOut(
    ctx context.Context,
    entrada <-chan Tarefa,
    numWorkers int,
) <-chan Resultado {
    resultados := make(chan Resultado, numWorkers*2)
    var wg sync.WaitGroup

    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(ctx, i, entrada, resultados, &wg)
    }

    // Fecha resultados quando todos os workers terminarem
    go func() {
        wg.Wait()
        close(resultados)
    }()

    return resultados
}

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

    // Alimenta o channel de tarefas
    tarefas := make(chan Tarefa, 10)
    go func() {
        defer close(tarefas)
        for i := 1; i <= 10; i++ {
            select {
            case tarefas <- Tarefa{ID: i, Dado: fmt.Sprintf("item_%d", i)}:
            case <-ctx.Done():
                return
            }
        }
    }()

    // 3 workers processam as 10 tarefas em paralelo
    resultados := fanOut(ctx, tarefas, 3)

    inicio := time.Now()
    for r := range resultados {
        fmt.Printf("[tarefa:%02d] %s (%.0fms)\n",
            r.TarefaID, r.Saida, r.Duracao.Seconds()*1000)
    }
    fmt.Printf("\ntotal: %v\n", time.Since(inicio))
}

Fan-in: consolidando resultados

Fan-in faz o oposto — combina múltiplos channels de entrada em um único channel de saída:

package main

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

// fanIn genérico — combina N channels em 1
func fanIn[T any](ctx context.Context, channels ...<-chan T) <-chan T {
    merged := make(chan T, len(channels))
    var wg sync.WaitGroup

    enviar := func(ch <-chan T) {
        defer wg.Done()
        for {
            select {
            case v, ok := <-ch:
                if !ok {
                    return
                }
                select {
                case merged <- v:
                case <-ctx.Done():
                    return
                }
            case <-ctx.Done():
                return
            }
        }
    }

    wg.Add(len(channels))
    for _, ch := range channels {
        go enviar(ch)
    }

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

    return merged
}

// Fonte de dados simulada
func fonte(ctx context.Context, nome string, intervalo time.Duration) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        ticker := time.NewTicker(intervalo)
        defer ticker.Stop()
        contador := 0
        for {
            select {
            case <-ticker.C:
                contador++
                select {
                case out <- fmt.Sprintf("[%s] evento_%d", nome, contador):
                case <-ctx.Done():
                    return
                }
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

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

    // Três fontes com cadências diferentes
    ch1 := fonte(ctx, "sensores", 200*time.Millisecond)
    ch2 := fonte(ctx, "logs", 150*time.Millisecond)
    ch3 := fonte(ctx, "metricas", 300*time.Millisecond)

    // Fan-in consolida tudo em um stream único
    todos := fanIn(ctx, ch1, ch2, ch3)

    for evento := range todos {
        fmt.Println(evento)
    }
}

A função fanIn usa generics — disponíveis desde Go 1.18 — tornando-a reutilizável para qualquer tipo sem duplicação.

Fan-out seguido de fan-in: o padrão completo

A combinação fan-out → processamento → fan-in é um dos padrões mais poderosos para processamento paralelo com ordem preservada opcional:

package main

import (
    "context"
    "fmt"
    "sort"
    "sync"
    "time"
)

type Item struct {
    ID    int
    Valor string
}

type ItemProcessado struct {
    ID        int
    Original  string
    Resultado string
}

func processarItem(ctx context.Context, item Item) (ItemProcessado, error) {
    select {
    case <-time.After(50 * time.Millisecond):
        return ItemProcessado{
            ID:        item.ID,
            Original:  item.Valor,
            Resultado: fmt.Sprintf("PROCESSADO(%s)", item.Valor),
        }, nil
    case <-ctx.Done():
        return ItemProcessado{}, ctx.Err()
    }
}

func processar(
    ctx context.Context,
    itens <-chan Item,
    erros chan<- error,
) <-chan ItemProcessado {
    out := make(chan ItemProcessado)
    go func() {
        defer close(out)
        for item := range itens {
            resultado, err := processarItem(ctx, item)
            if err != nil {
                select {
                case erros <- err:
                case <-ctx.Done():
                }
                return
            }
            select {
            case out <- resultado:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func processarParalelo(
    ctx context.Context,
    itens []Item,
    concorrencia int,
) ([]ItemProcessado, error) {
    entrada := make(chan Item, len(itens))
    for _, item := range itens {
        entrada <- item
    }
    close(entrada)

    erros := make(chan error, concorrencia)

    // Fan-out: N goroutines de processamento
    channels := make([]<-chan ItemProcessado, concorrencia)
    for i := 0; i < concorrencia; i++ {
        channels[i] = processar(ctx, entrada, erros)
    }

    // Fan-in: consolida resultados
    var wg sync.WaitGroup
    merged := make(chan ItemProcessado, len(itens))

    for _, ch := range channels {
        wg.Add(1)
        ch := ch
        go func() {
            defer wg.Done()
            for v := range ch {
                merged <- v
            }
        }()
    }

    go func() {
        wg.Wait()
        close(merged)
        close(erros)
    }()

    // Coleta resultados
    var resultados []ItemProcessado
    for r := range merged {
        resultados = append(resultados, r)
    }

    // Verifica erros
    if err, ok := <-erros; ok {
        return nil, err
    }

    // Reordena por ID para resultado determinístico
    sort.Slice(resultados, func(i, j int) bool {
        return resultados[i].ID < resultados[j].ID
    })

    return resultados, nil
}

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

    itens := make([]Item, 12)
    for i := range itens {
        itens[i] = Item{ID: i + 1, Valor: fmt.Sprintf("item_%02d", i+1)}
    }

    inicio := time.Now()
    resultados, err := processarParalelo(ctx, itens, 4)
    if err != nil {
        fmt.Println("Erro:", err)
        return
    }

    for _, r := range resultados {
        fmt.Printf("[%02d] %s → %s\n", r.ID, r.Original, r.Resultado)
    }
    fmt.Printf("\n12 itens em %v (4 workers)\n", time.Since(inicio))
}

Esse encadeamento — distribuir, recolher, parar tudo no primeiro erro — é tão comum que ganhou uma biblioteca semioficial, o golang.org/x/sync/errgroup. Com g, ctx := errgroup.WithContext(ctx), cada tarefa é disparada com g.Go(func() error { ... }), g.SetLimit(4) limita quantas rodam ao mesmo tempo, o primeiro erro cancela o ctx das demais, e g.Wait() devolve esse erro. Gravações em posições fixas de uma fatia pré-alocada, uma por tarefa, dispensam o canal de saída e a reordenação. Vale conhecer a versão manual, acima, para entender o que a biblioteca faz.

Or-done: lendo de channel com cancelamento

Um utilitário elegante que encapsula a leitura de um channel com suporte a cancelamento via context:

func orDone[T any](ctx context.Context, ch <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            select {
            case v, ok := <-ch:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Uso
func main() {
    ctx, cancelar := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancelar()

    ch := make(chan int)
    go func() {
        for i := 0; ; i++ {
            ch <- i
            time.Sleep(100 * time.Millisecond)
        }
    }()

    for v := range orDone(ctx, ch) {
        fmt.Println(v)
    }
    fmt.Println("encerrado por timeout")
}

Batch processing: processamento em lotes

Processar itens individualmente pode ser ineficiente quando o custo por operação é alto — por exemplo, inserções em banco de dados. O padrão de batch agrupa itens e os processa juntos:

package main

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

func batcher[T any](
    ctx context.Context,
    entrada <-chan T,
    tamanho int,
    espera time.Duration,
) <-chan []T {
    out := make(chan []T)

    go func() {
        defer close(out)
        lote := make([]T, 0, tamanho)
        timer := time.NewTimer(espera)
        defer timer.Stop()

        enviarLote := func() {
            if len(lote) > 0 {
                copia := make([]T, len(lote))
                copy(copia, lote)
                select {
                case out <- copia:
                case <-ctx.Done():
                    return
                }
                lote = lote[:0]
            }
            if !timer.Stop() {
                select {
                case <-timer.C:
                default:
                }
            }
            timer.Reset(espera)
        }

        for {
            select {
            case item, ok := <-entrada:
                if !ok {
                    enviarLote()
                    return
                }
                lote = append(lote, item)
                if len(lote) >= tamanho {
                    enviarLote()
                }
            case <-timer.C:
                enviarLote() // envia lote parcial por timeout
            case <-ctx.Done():
                return
            }
        }
    }()

    return out
}

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

    entrada := make(chan int)
    go func() {
        defer close(entrada)
        for i := 1; i <= 17; i++ {
            select {
            case entrada <- i:
                time.Sleep(80 * time.Millisecond)
            case <-ctx.Done():
                return
            }
        }
    }()

    // Lotes de até 5 itens, ou a cada 300ms
    lotes := batcher(ctx, entrada, 5, 300*time.Millisecond)

    for lote := range lotes {
        fmt.Printf("lote de %d: %v\n", len(lote), lote)
    }
}

Semáforo: limitando concorrência

Quando é necessário limitar o número de goroutines simultâneas — por exemplo, para não sobrecarregar uma API externa com chamadas paralelas — um channel com buffer funciona como semáforo:

package main

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

type Semaforo chan struct{}

func NovoSemaforo(limite int) Semaforo {
    return make(Semaforo, limite)
}

func (s Semaforo) Adquirir(ctx context.Context) error {
    select {
    case s <- struct{}{}:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (s Semaforo) Liberar() {
    <-s
}

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

    sem := NovoSemaforo(3) // máximo 3 operações simultâneas
    var wg sync.WaitGroup

    for i := 1; i <= 10; i++ {
        wg.Add(1)
        i := i
        go func() {
            defer wg.Done()

            if err := sem.Adquirir(ctx); err != nil {
                fmt.Printf("tarefa %d cancelada\n", i)
                return
            }
            defer sem.Liberar()

            fmt.Printf("tarefa %d iniciada (ativas: %d)\n", i, len(sem))
            time.Sleep(200 * time.Millisecond)
            fmt.Printf("tarefa %d concluída\n", i)
        }()
    }

    wg.Wait()
}

Para limites com pesos diferentes — uma tarefa grande que ocupa três vagas, outra pequena que ocupa uma —, o pacote golang.org/x/sync/semaphore oferece o semaphore.Weighted, com Acquire(ctx, n) e Release(n). Para o caso simples de limitar goroutines, o canal com buffer do exemplo, ou o SetLimit do errgroup, costumam bastar.

Pipelines, fan-out, fan-in, lotes e semáforos são combinações das mesmas poucas peças — goroutines, canais, select e context —, e é justamente por isso que viram vocabulário comum entre quem escreve Go. O que separa a versão de exemplo da versão de produção é quase sempre o mesmo detalhe: todo envio que pode ficar sem receptor precisa de uma saída pelo contexto, e quem desiste de ler precisa cancelar para avisar quem escreve. Com essa regra seguida em cada estágio, e o errgroup para o caso mais comum, os padrões param de vazar goroutines.

Fontes e leituras recomendadas

• Go Blog: Go Concurrency Patterns: Pipelines and cancellation — Artigo oficial sobre pipelines com cancelamento. https://go.dev/blog/pipelines

• Go Blog: Advanced Go Concurrency Patterns — Padrões avançados apresentados na GopherCon. https://go.dev/blog/advanced-go-concurrency-patterns

• Concurrency in Go — Katherine Cox-Buday — Livro dedicado inteiramente a padrões de concorrência em Go. https://www.oreilly.com/library/view/concurrency-in-go/9781491941294/

• Go by Example: Worker Pools — Exemplo prático de worker pool com channels. https://gobyexample.com/worker-pools

• golang.org/x/sync — semaphore — Implementação oficial de semáforo ponderado. https://pkg.go.dev/golang.org/x/sync/semaphore

• golang.org/x/sync — errgroup — WaitGroup com propagação de erros, essencial para pipelines. https://pkg.go.dev/golang.org/x/sync/errgroup

Exercícios

Exercício 1

No exemplo de or-done, o programa imprime "encerrado por timeout" e parece terminar bem. Um colega usa o mesmo padrão dentro de um servidor, com uma goroutine produtora por requisição, e o número de goroutines não para de crescer. O que o or-done protege, o que ele não protege, e como corrigir o produtor?

Exercício 2

No batcher do artigo, a função enviarLote copia o lote para uma fatia nova antes de enviá-lo, e só depois faz lote = lote[:0]. Um revisor acha a cópia um desperdício e propõe enviar lote direto. O que o consumidor passaria a receber?

Exercício 3

No exemplo de fan-out, cada worker envia o resultado com resultados <- Resultado{...}, sem select. O consumidor é alterado para parar de ler e retornar assim que encontra o primeiro resultado com erro. Que problema isso cria, e quais são as duas formas de resolvê-lo?

Exercício 4

Um fan-out com 8 workers processa uma lista de pedidos e grava os resultados num canal, na ordem em que ficam prontos. O relatório final precisa sair na ordem original. O artigo reordena com sort.Slice pelo ID no final. Descreva duas alternativas que dispensam a ordenação e diga quando cada uma é melhor.

Exercício 5

Um serviço chama uma API parceira que aceita no máximo 10 requisições simultâneas por cliente. Para respeitar o limite, o time usa o semáforo do artigo com capacidade 10, e mesmo assim recebe erros 429 (Too Many Requests) em picos. O semáforo está errado? Que outra forma de limite pode estar em jogo?