Worker pool em Go: limitando goroutines, resultados e cancelamento

1. Visão geral

Goroutines em Go são muito baratas, então chamar go para cada tarefa é comum. Quando o número de tarefas atinge dezenas de milhares e cada tarefa chama banco de dados ou HTTP, essa abordagem facilmente esgota o connection pool ou ultrapassa o rate limit do serviço externo.

Este artigo te guia na construção de um worker pool, ou seja, um número fixo de goroutines que recebem tarefas de um channel. Ao final, você saberá como escrever um pool básico, coletar resultados e erros, cancelar com context, simplificar com errgroup.SetLimit e testar que o pool realmente limita o paralelismo.

2. Problema: goroutines ilimitadas

Considere este código que processa uma lista de pedidos:

for _, o := range orders {
	go chargeOrder(o) // 50.000 pedidos -> 50.000 goroutines chamando payment API simultaneamente
}

Cada goroutine consome apenas alguns KB de memória, então o problema geralmente não está na RAM. O problema está nos recursos downstream:

• O connection pool do banco de dados tem apenas cerca de 20–50 conexões.
• O gateway de pagamento limita o número de requisições por segundo.
• O número de file descriptors do processo é limitado.

O worker pool resolve isso separando dois conceitos. O número de tarefas pode ser muito grande, enquanto o número de tarefas executando simultaneamente é limitado pelo número de workers.

3. Worker pool básico

O modelo tem três partes:

1. Um channel jobs contendo tarefas a fazer.
2. N goroutines worker, cada worker faz um loop for range no channel.
3. Um sync.WaitGroup para esperar todos os workers terminarem.

package main

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

type Order struct {
	ID     int
	Amount float64
}

func worker(id int, orders <-chan Order, wg *sync.WaitGroup) {
	defer wg.Done()
	for o := range orders {
		time.Sleep(100 * time.Millisecond) // simulando chamada a payment API
		fmt.Printf("worker %d processando order %d (%.0f)\n", id, o.ID, o.Amount)
	}
}

func main() {
	orders := make(chan Order)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, orders, &wg)
	}

	start := time.Now()
	for i := 1; i <= 9; i++ {
		orders <- Order{ID: i, Amount: float64(i) * 10}
	}
	close(orders)
	wg.Wait()
	fmt.Println("concluído em", time.Since(start).Round(100*time.Millisecond))
}

O código funciona assim:

• orders <-chan Order é um channel somente para recebimento no lado do worker. O compilador bloqueará o worker se tentar enviar ou fechar o channel.
• for o := range orders sai automaticamente quando o channel é fechado (close) e todos os elementos são esgotados.
• close(orders) é o sinal de “fim do trabalho”. Se você esquecer dessa linha, o worker fica esperando para sempre e wg.Wait() causa deadlock.
• wg.Add(1) deve ser chamado antes de go worker(...), não dentro da goroutine.

Saída da execução (a ordem das linhas pode variar):

worker 3 processando order 1 (10)
worker 1 processando order 2 (20)
worker 2 processando order 3 (30)
worker 2 processando order 4 (40)
...
concluído em 300ms

9 pedidos divididos entre 3 workers, cada pedido leva 100ms, então o tempo total é cerca de 300ms em vez de 900ms se fosse sequencial.

Nota: A partir do Go 1.25, você pode escrever wg.Go(func() { worker(i, orders) }). Essa função chama automaticamente Add e Done, evitando o erro de esquecer um dos dois.

4. Coletando resultados e erros

Na prática, você geralmente precisa saber qual pedido foi bem-sucedido e qual falhou. Adicionamos um channel results para isso:

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Order struct {
	ID     int
	Amount float64
}

type Result struct {
	OrderID int
	Fee     float64
	Err     error
}

func chargeOrder(o Order) (float64, error) {
	if o.Amount <= 0 {
		return 0, errors.New("valor inválido")
	}
	return o.Amount * 0.02, nil // taxa de 2%
}

func main() {
	orders := []Order{{1, 100}, {2, 0}, {3, 250}, {4, 80}}

	jobs := make(chan Order)
	results := make(chan Result)
	var wg sync.WaitGroup

	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				fee, err := chargeOrder(o)
				results <- Result{OrderID: o.ID, Fee: fee, Err: err}
			}
		}()
	}

	go func() {
		for _, o := range orders {
			jobs <- o
		}
		close(jobs)
	}()

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

	for r := range results {
		if r.Err != nil {
			fmt.Printf("order %d erro: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("order %d taxa %.2f\n", r.OrderID, r.Fee)
	}
}

A saída tem a forma:

order 1 taxa 2.00
order 3 taxa 5.00
order 2 erro: valor inválido
order 4 taxa 1.60

Há três pontos a notar:

• O envio de jobs executado em uma goroutine separada. Se main enviasse jobs e lesse resultados na mesma goroutine, o worker ficaria bloqueado em results <- enquanto main ficaria bloqueado em jobs <-. O resultado seria um deadlock.
• A terceira goroutine espera por wg.Wait() antes de fazer close(results). Só quando todos os workers terminarem podemos garantir que ninguém mais está enviando para results. Regra geral: apenas o lado que envia fecha o channel.
• Erros são encapsulados em Result em vez de parar todo o pool. O erro de um pedido não deve bloquear outros pedidos.

A ordem dos resultados não corresponde à ordem de entrada. Se você precisar manter a ordem, leve o índice (index) junto com o job, ou use a abordagem na seção 6.

5. Cancelando o pool com context

Quando uma requisição HTTP é cancelada ou atinge o timeout, o pool deve parar em vez de continuar chamando a payment API. Você faz isso deixando o worker escutar ctx.Done():

func worker(ctx context.Context, jobs <-chan Order, results chan<- Result) {
	for {
		select {
		case <-ctx.Done():
			return
		case o, ok := <-jobs:
			if !ok {
				return
			}
			fee, err := chargeOrder(o)
			select {
			case results <- Result{OrderID: o.ID, Fee: fee, Err: err}:
			case <-ctx.Done():
				return
			}
		}
	}
}

O lado producer também precisa de um select similar:

go func() {
	defer close(jobs)
	for _, o := range orders {
		select {
		case jobs <- o:
		case <-ctx.Done():
			return
		}
	}
}()

E crie um context com timeout no ponto de chamada:

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

A razão pela qual você precisa de select em ambas as direções de envio: se o worker apenas verificar ctx ao receber um job, ele ainda pode ficar preso em results <- quando o lado leitor já desistiu. Esta é uma fonte muito comum de vazamento de goroutine (goroutine leak).

Nota: select apenas ajuda o worker a sair entre jobs. Se chargeOrder fizer uma chamada HTTP, você precisa passar ctx para ela (http.NewRequestWithContext) para poder cancelar também a requisição em andamento.

6. Simplificando com errgroup.SetLimit

Escrever channel, WaitGroup e context manualmente é fácil de errar. O pacote errgroup tem SetLimit, permitindo criar um pool limitado em apenas algumas linhas. Primeiro, adicione a dependência:

go get golang.org/x/sync/errgroup

A função genérica abaixo processa um slice com no máximo limit goroutines, mantém a ordem dos resultados e para no primeiro erro:

package pool

import (
	"context"

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

func Process[T, R any](
	ctx context.Context,
	limit int,
	items []T,
	fn func(context.Context, T) (R, error),
) ([]R, error) {
	results := make([]R, len(items))
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for i, item := range items {
		g.Go(func() error {
			r, err := fn(ctx, item)
			if err != nil {
				return err
			}
			results[i] = r
			return nil
		})
	}

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

O código funciona assim:

• g.SetLimit(limit) faz com que g.Go seja bloqueado quando já há limit goroutines em execução. Em termos de eficiência, é equivalente a um worker pool.
• errgroup.WithContext retorna um ctx que será cancelado assim que qualquer função retornar um erro. Outras tarefas usando ctx receberão o sinal para parar.
• results[i] = r é seguro durante a execução concorrente porque cada goroutine escreve em um índice único. Não é necessário mutex neste caso.
• A partir do Go 1.22, as variáveis de loop i e item são criadas novas a cada iteração, então closures não compartilham mais as mesmas variáveis como em versões antigas.

Forma de chamar:

fees, err := pool.Process(ctx, 5, orders, func(ctx context.Context, o Order) (float64, error) {
	return paymentService.Charge(ctx, o)
})

Diferença em relação a um pool manual: errgroup cria uma nova goroutine para cada item em vez de reutilizar N workers. Para a maioria das aplicações, essa diferença é negligenciável. Se você precisa que workers mantenham estado próprio, como cada worker tendo sua própria conexão, use o pool manual das seções 3–5.

7. Testando o worker pool

Você precisa testar dois comportamentos principais. Primeiro, o pool nunca ultrapassa o limit. Segundo, erros são retornados corretamente. Para medir o nível de paralelismo, usamos atomic.Int32:

package pool

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestProcess_LimitsConcurrencyAndKeepsOrder(t *testing.T) {
	var running, maxSeen atomic.Int32
	items := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	got, err := Process(context.Background(), 3, items, func(_ context.Context, n int) (int, error) {
		cur := running.Add(1)
		defer running.Add(-1)
		for {
			m := maxSeen.Load()
			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return n * 2, nil
	})

	if err != nil {
		t.Fatalf("não esperava erro: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("concorrência de %d, limite é 3", maxSeen.Load())
	}
	want := []int{2, 4, 6, 8, 10, 12, 14, 16, 18, 20}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("got[%d] = %d, want %d", i, got[i], want[i])
		}
	}
}

func TestProcess_ReturnsFirstError(t *testing.T) {
	errDeclined := errors.New("cartão recusado")

	_, err := Process(context.Background(), 2, []int{1, 2, 3}, func(_ context.Context, n int) (int, error) {
		if n == 2 {
			return 0, errDeclined
		}
		return n, nil
	})

	if !errors.Is(err, errDeclined) {
		t.Fatalf("espera errDeclined, recebeu %v", err)
	}
}

Execute o teste com race detector:

go test -race ./...

ok  	example.com/shop/pool	0.061s

O loop CompareAndSwap atualiza maxSeen sem necessidade de lock. A flag -race é obrigatória para código concorrente, pois detecta acessos a dados compartilhados que os olhos facilmente perdem.

8. Erros comuns e quando não usar

Erros comuns:

• Esquecer close(jobs): o worker nunca sai e o programa relata fatal error: all goroutines are asleep - deadlock!.
• Fechar o channel no lado receptor, ou fechar duas vezes: causa panic: close of closed channel.
• Chamar wg.Add dentro de uma goroutine: Wait pode retornar antes que a goroutine consiga fazer Add.
• Enviar jobs e ler resultados na mesma goroutine com channel sem buffer: causa deadlock como analisado na seção 4.
• Não fazer select em ctx.Done() ao enviar resultados: leva a goroutine leak quando o lado leitor já saiu cedo.
• Escolher número de workers arbitrariamente: para tarefas CPU-bound, use runtime.NumCPU(). Para tarefas I/O-bound, baseie-se no limite do recurso downstream, por exemplo o tamanho do connection pool.

Quando não usar worker pool:

• Número de tarefas pequeno e conhecido de antemão, como chamar 3 serviços em paralelo. Neste caso, errgroup sem SetLimit é suficiente.
• Tarefas muito curtas, apenas alguns micro segundos. O custo de passar pelo channel pode ser maior que o próprio trabalho, então processamento sequencial ou em lotes (batches) será mais eficiente.
• Você precisa limitar por taxa (requisições/segundo) e não por número de execuções simultâneas. Neste caso, você precisa de um rate limiter, como golang.org/x/time/rate. Worker pool não substitui um rate limiter.

9. Conclusão

O worker pool ajuda você a controlar o nível de paralelismo em Go: o número de tarefas pode ser muito grande, mas os recursos downstream ainda são protegidos. Este artigo passou pelos seguintes passos:

• Pool básico com channel e sync.WaitGroup.
• Coletar resultados e erros via channel results, fechar o channel no lugar certo.
• Cancelamento seguro com context em ambas as direções de envio.
• Simplificar com errgroup.SetLimit usando uma função Process genérica.
• Testar limite de concorrência e erros com go test -race.

Para a maioria dos casos, errgroup.SetLimit é a escolha mais simples e menos propensa a erros. Você deve implementar um pool manualmente apenas quando workers precisam manter seu próprio estado ou quando precisa de controle fino sobre o fluxo de jobs e resultados.