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

Os artigos anteriores 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 20 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 pares → 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))
}

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

Resumo do que foi coberto

Este artigo apresentou os padrões arquiteturais de concorrência em Go: pipelines com cancelamento via context, fan-out para distribuição de trabalho em múltiplos workers, fan-in para consolidação de múltiplos channels, a combinação fan-out/fan-in para processamento paralelo com reordenação, o utilitário or-done para leitura segura com cancelamento, batch processing para agrupamento de itens e semáforos para limitação de concorrência. O próximo e último artigo do módulo aborda as armadilhas da concorrência e como detectá-las.

Referências e leituras complementares

• 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