# Padrões de Concorrência

Lá nos dias 17 a 20 vimos as peças básicas de concorrência em Go: goroutines, WaitGroup, Mutex e channels. Com essas peças conseguimos fazer praticamente qualquer coisa, mas alguns problemas aparecem com tanta frequência que já existem “receitas” conhecidas para resolvê-los. Hoje vamos ver três delas: worker pool, fan-out/fan-in e o errgroup.

# Worker Pool

Imagine que temos 10.000 imagens para redimensionar. Criar uma goroutine para cada imagem é fácil, goroutines são baratas, mas cada uma vai abrir um arquivo, alocar memória, usar CPU… Com 10.000 rodando ao mesmo tempo podemos acabar com a memória da máquina, estourar o limite de arquivos abertos ou derrubar uma API externa com requisições demais.

O worker pool resolve isso criando um número fixo de goroutines (os workers) que ficam pegando tarefas de uma fila. A fila é um channel:

package main

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

func worker(id int, jobs <-chan int, resultados chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()
	for job := range jobs {
		time.Sleep(500 * time.Millisecond) // simulando um processamento pesado
		resultados <- fmt.Sprintf("worker %d processou o job %d", id, job)
	}
}

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

	jobs := make(chan int, numJobs)
	resultados := make(chan string, numJobs)

	var wg sync.WaitGroup
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go worker(i, jobs, resultados, &wg)
	}

	inicio := time.Now()

	for j := 1; j <= numJobs; j++ {
		jobs <- j
	}
	close(jobs)

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

	for r := range resultados {
		fmt.Println(r)
	}

	fmt.Printf("tempo total: %v\n", time.Since(inicio).Round(time.Millisecond))
}

Saída:

worker 2 processou o job 3
worker 1 processou o job 2
worker 3 processou o job 1
worker 3 processou o job 6
worker 1 processou o job 5
worker 2 processou o job 4
worker 3 processou o job 7
worker 1 processou o job 8
worker 2 processou o job 9
tempo total: 1.502s

São 9 jobs de 500ms cada. Sequencialmente levaria 4,5 segundos; com 3 workers levou 1,5 segundo, e nunca tivemos mais de 3 jobs rodando ao mesmo tempo.

Alguns detalhes importantes:

• Os tipos dos channels nos parâmetros: jobs <-chan int só permite receber e resultados chan<- string só permite enviar. O compilador garante que o worker não vai, por exemplo, escrever na fila de jobs por engano.
• O close(jobs): o for job := range jobs só termina quando o channel é fechado. Sem o close, os workers ficariam esperando novos jobs para sempre.
• Quem fecha o resultados: não podemos fechar logo depois de enviar os jobs, porque os workers ainda estão escrevendo nele. Por isso uma goroutine separada espera todos os workers terminarem (wg.Wait()) e só então fecha o channel. Isso libera o for r := range resultados do main.

# Fan-out e Fan-in

Esses dois nomes descrevem o fluxo dos dados:

• Fan-out: várias goroutines lendo do mesmo channel, dividindo o trabalho entre elas. O worker pool acima já é um fan-out.
• Fan-in: juntar vários channels em um só.

Esse padrão aparece muito em pipelines, onde cada etapa é uma função que recebe um channel e devolve outro:

package main

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

// gerador: transforma uma lista de números num channel
func gerador(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}

// quadrado: lê números de um channel e devolve o quadrado em outro
func quadrado(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			time.Sleep(300 * time.Millisecond) // simulando um cálculo lento
			out <- n * n
		}
	}()
	return out
}

// merge: junta vários channels em um só (fan-in)
func merge(channels ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup

	for _, c := range channels {
		wg.Add(1)
		go func(c <-chan int) {
			defer wg.Done()
			for n := range c {
				out <- n
			}
		}(c)
	}

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

	return out
}

func main() {
	inicio := time.Now()

	entrada := gerador(1, 2, 3, 4, 5, 6)

	// fan-out: três goroutines lendo do mesmo channel
	c1 := quadrado(entrada)
	c2 := quadrado(entrada)
	c3 := quadrado(entrada)

	// fan-in: juntando os resultados
	soma := 0
	for n := range merge(c1, c2, c3) {
		fmt.Println(n)
		soma += n
	}

	fmt.Println("soma:", soma)
	fmt.Printf("tempo total: %v\n", time.Since(inicio).Round(time.Millisecond))
}

Saída:

4
9
1
36
16
25
soma: 91
tempo total: 602ms

O fluxo fica assim:

                  ┌─> quadrado (c1) ─┐
gerador(entrada) ─┼─> quadrado (c2) ─┼─> merge ─> main
                  └─> quadrado (c3) ─┘
       fan-out                         fan-in

Seis cálculos de 300ms levariam 1,8 segundo em sequência; com três goroutines levou 600ms. Repare que os resultados chegam fora de ordem: quem termina primeiro entrega primeiro. Se a ordem importar, é preciso enviar junto um índice e reordenar no final.

O legal desse formato é que cada etapa é independente e conversa com as outras só por channels. Dá para encaixar novas etapas (filtrar, salvar no banco…) sem mexer nas que já existem.

# errgroup

Nos exemplos acima nenhuma goroutine retorna erro. No mundo real, quase sempre retorna: a requisição falhou, o arquivo não existe, o banco caiu… Com WaitGroup teríamos que criar um channel ou um slice protegido por mutex só para coletar os erros, e ainda cancelar as outras goroutines manualmente quando uma falhasse.

O pacote errgroup faz tudo isso. Ele não está na biblioteca padrão, mas é mantido pelo time do Go:

go get golang.org/x/sync

Exemplo buscando 5 usuários em paralelo, onde a busca do usuário 3 falha:

package main

import (
	"context"
	"fmt"
	"time"

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

func buscarUsuario(ctx context.Context, id int) error {
	duracao := time.Duration(id) * 300 * time.Millisecond

	select {
	case <-time.After(duracao):
		if id == 3 {
			return fmt.Errorf("usuário %d não encontrado", id)
		}
		fmt.Printf("usuário %d carregado\n", id)
		return nil
	case <-ctx.Done():
		fmt.Printf("busca do usuário %d cancelada\n", id)
		return ctx.Err()
	}
}

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

	for id := 1; id <= 5; id++ {
		g.Go(func() error {
			return buscarUsuario(ctx, id)
		})
	}

	if err := g.Wait(); err != nil {
		fmt.Println("erro:", err)
	}
}

Saída:

usuário 1 carregado
usuário 2 carregado
busca do usuário 5 cancelada
busca do usuário 4 cancelada
erro: usuário 3 não encontrado

O que aconteceu:

• g.Go substitui o wg.Add(1) + go func() + defer wg.Done(), e a função passada retorna um error.
• g.Wait() espera todas as goroutines e retorna o primeiro erro que aconteceu.
• O errgroup.WithContext retorna um context que é cancelado assim que alguma goroutine retorna erro. Os usuários 4 e 5 ainda estavam sendo buscados quando o 3 falhou, e pararam na hora, sem desperdiçar trabalho. É o context do dia 89 em ação.

Usamos o id dentro da closure sem copiar a variável. Desde o Go 1.22 cada iteração do for tem sua própria variável, então isso é seguro. Em versões antigas seria preciso fazer id := id antes do g.Go.

## SetLimit: um worker pool em poucas linhas

O errgroup também tem o SetLimit, que limita quantas goroutines rodam ao mesmo tempo. Quando o limite é atingido, o g.Go bloqueia até uma vaga abrir. Na prática é um worker pool sem precisar criar os channels:

package main

import (
	"fmt"
	"sync/atomic"
	"time"

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

func main() {
	var g errgroup.Group
	g.SetLimit(3)

	var ativas atomic.Int32
	inicio := time.Now()

	for job := 1; job <= 9; job++ {
		g.Go(func() error {
			n := ativas.Add(1)
			fmt.Printf("job %d começou (%d rodando)\n", job, n)
			time.Sleep(500 * time.Millisecond)
			ativas.Add(-1)
			return nil
		})
	}

	g.Wait()
	fmt.Printf("tempo total: %v\n", time.Since(inicio).Round(time.Millisecond))
}

Saída:

job 1 começou (1 rodando)
job 3 começou (3 rodando)
job 2 começou (2 rodando)
job 4 começou (1 rodando)
job 6 começou (2 rodando)
job 5 começou (3 rodando)
job 7 começou (1 rodando)
job 8 começou (2 rodando)
job 9 começou (3 rodando)
tempo total: 1.503s

O mesmo resultado do primeiro exemplo: 9 jobs, nunca mais que 3 ao mesmo tempo, 1,5 segundo no total. O atomic.Int32 é só para contar com segurança quantas goroutines estão ativas, sem precisar de um mutex.

# E o WaitGroup.Go?

Falando em diminuir código repetido: a partir do Go 1.25 o próprio sync.WaitGroup ganhou um método Go, que faz o Add(1) e o Done() por nós:

var wg sync.WaitGroup
for i := 1; i <= 3; i++ {
	wg.Go(func() {
		fmt.Println("goroutine", i)
	})
}
wg.Wait()

Ele não trata erros nem cancela nada, mas para os casos simples já evita esquecer um wg.Add ou um defer wg.Done().

# Qual usar?

• Só quero rodar várias coisas e esperar: sync.WaitGroup (com wg.Go se estiver no Go 1.25+).
• As tarefas podem falhar e, se uma falhar, as outras podem parar: errgroup.WithContext.
• Preciso limitar a quantidade de tarefas simultâneas: errgroup com SetLimit, ou um worker pool com channels se precisar de mais controle (workers com estado, fila que recebe jobs continuamente…).
• Os dados passam por várias etapas: pipeline com fan-out/fan-in.

# Referências

1. Go Concurrency Patterns: Pipelines and cancellation
2. Go by Example: Worker Pools
3. Documentação errgroup
4. Documentação sync.WaitGroup