Worker Pools e Rate Limiting em Go: Concorrência Controlada e Vazão

Em Go, disparar uma goroutine custa apenas ~2KB de memória inicial, o que frequentemente seduz desenvolvedores iniciantes a adotar o padrão ingênuo de criar uma goroutine para cada tarefa recebida (for _, item := range items { go process(item) }).

No entanto, em ambientes de produção, recursos externos não são infinitos. Abrir 50.000 goroutines simultâneas que tentam se conectar ao banco de dados database-sql e Connection Pooling ou chamar uma API externa resulta em esgotamento de descritores de arquivos do sistema operacional (too many open files), saturação do pool de conexões e travamento da máquina por falta de memória (OOM).

Para manter sistemas estáveis sob cargas massivas, combinamos dois padrões clássicos de concorrência: Worker Pools (para fixar o teto de concorrência interna) e Rate Limiting (para controlar a taxa de transferência no tempo).

---

1. O Padrão Worker Pool

Um Worker Pool estabelece um número pré-determinado de goroutines trabalhadoras ($N$) que consomem tarefas concorrentemente a partir de uma fila de entrada comum (canal Go), processam os dados e entregam as respostas em um canal de resultados.

Mecânica de Funcionamento:

1. Canal de Tarefas (jobs): Um canal bufferizado onde o produtor envia os itens a serem processados.
2. Pool de Workers: $N$ goroutines iniciadas no startup que ficam em loop esperando mensagens do canal jobs.
3. Distribuição Equilibrada: O próprio runtime do Go encarrega-se de entregar tarefas para o primeiro worker que estiver desocupado (work stealing implícito nos canais).
4. Sincronização: Um sync.WaitGroup garante que todos os workers concluam suas tarefas pendentes antes do canal de resultados ser fechado.

---

Implementação Robusta com Context e Cancelamento

package workerpool

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

type Job struct {
	ID    int
	Input string
}

type Result struct {
	JobID  int
	Output string
	Err    error
}

// StartWorkerPool gerencia um pool fixo de workers resilientes a cancelamento
func StartWorkerPool(ctx context.Context, numWorkers int, jobs <-chan Job, results chan<- Result) {
	var wg sync.WaitGroup

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

			for {
				select {
				case <-ctx.Done():
					// Interrompe imediatamente se o contexto for cancelado (ex: shutdown)
					return
				case job, ok := <-jobs:
					if !ok {
						// Canal de jobs fechado pelo produtor
						return
					}

					// Processa a tarefa
					res := processJob(ctx, job)

					// Envia resultado respeitando possível cancelamento
					select {
					case <-ctx.Done():
						return
					case results <- res:
					}
				}
			}
		}(i)
	}

	// Aguarda todos os workers terminarem antes de fechar o canal de resultados
	wg.Wait()
	close(results)
}

func processJob(ctx context.Context, job Job) Result {
	// Simula processamento pesado
	return Result{
		JobID:  job.ID,
		Output: fmt.Sprintf("Processado: %s", job.Input),
	}
}

---

2. O Padrão Rate Limiting e o Algoritmo Token Bucket

Enquanto o Worker Pool limita a quantidade de tarefas simultâneas, o Rate Limiting limita a velocidade com que as operações podem ser admitidas em uma janela de tempo (ex: no máximo 100 requisições por segundo).

O algoritmo padrão da indústria é o Token Bucket (implementado no pacote oficial golang.org/x/time/rate):

Como o Token Bucket Opera:

1. O balde possui uma capacidade máxima fixa de $b$ tokens (Burst).
2. O balde é reabastecido continuamente a uma taxa fixa de $r$ tokens por segundo (Rate).
3. Cada requisição exige $1$ token para ser processada:
   • Se houver token disponível no balde, ele é consumido e a requisição prossegue imediatamente (Allow() = true).
   • Se o balde estiver vazio, a requisição é rejeitada com código HTTP 429 Too Many Requests (ou aguarda com Wait(ctx)).

---

3. Rate Limiting de Clientes em APIs HTTP (Por Endereço IP)

Em uma API pública ou de microsserviços, aplicar um limitador global único penaliza todos os usuários se um único cliente malicioso disparar um ataque. A prática padrão é manter limitadores isolados por IP ou por API Key:

package middleware

import (
	"net/http"
	"sync"
	"time"

	"golang.org/x/time/rate"
)

type clientLimiter struct {
	limiter  *rate.Limiter
	lastSeen time.Time
}

type IPRateLimiter struct {
	mu      sync.Mutex
	clients map[string]*clientLimiter
	rate    rate.Limit
	burst   int
}

func NewIPRateLimiter(r rate.Limit, b int) *IPRateLimiter {
	lim := &IPRateLimiter{
		clients: make(map[string]*clientLimiter),
		rate:    r,
		burst:   b,
	}

	// Goroutine de limpeza periódica para evitar vazamento de memória com IPs antigos
	go lim.cleanupClients(time.Minute * 5)
	return lim
}

func (i *IPRateLimiter) getClient(ip string) *rate.Limiter {
	i.mu.Lock()
	defer i.mu.Unlock()

	client, exists := i.clients[ip]
	if !exists {
		client = &clientLimiter{
			limiter: rate.NewLimiter(i.rate, i.burst),
		}
		i.clients[ip] = client
	}

	client.lastSeen = time.Now()
	return client.limiter
}

func (i *IPRateLimiter) cleanupClients(ttl time.Duration) {
	for {
		time.Sleep(ttl)
		i.mu.Lock()
		for ip, client := range i.clients {
			if time.Since(client.lastSeen) > ttl {
				delete(i.clients, ip)
			}
		}
		i.mu.Unlock()
	}
}

// Middleware HTTP que bloqueia requisições excessivas com 429
func (i *IPRateLimiter) Middleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ip := r.RemoteAddr // Em produção com proxy reverso, extraia do X-Forwarded-For

		limiter := i.getClient(ip)
		if !limiter.Allow() {
			w.Header().Set("Retry-After", "1")
			http.Error(w, `{"error":"rate_limit_exceeded"}`, http.StatusTooManyRequests)
			return
		}

		next.ServeHTTP(w, r)
	})
}

---

4. Rate Limiting de Saída: Consumindo APIs com Quotas Rígidas

Ao consumir serviços externos de terceiros (como OpenAI, Stripe ou Discord), o seu cliente Go deve respeitar a quota do provedor para evitar bloqueios de conta.

O método limiter.Wait(ctx) bloqueia a goroutine até que um token esteja disponível ou até o contexto expirar:

package client

import (
	"context"
	"fmt"
	"net/http"

	"golang.org/x/time/rate"
)

type ExternalAPIClient struct {
	limiter *rate.Limiter
	client  *http.Client
}

func NewExternalAPIClient() *ExternalAPIClient {
	// Limita a 5 requisições por segundo, com burst de 1
	return &ExternalAPIClient{
		limiter: rate.NewLimiter(rate.Limit(5), 1),
		client:  &http.Client{},
	}
}

func (c *ExternalAPIClient) FetchData(ctx context.Context, endpoint string) error {
	// Aguarda o próximo token disponível de forma limpa
	if err := c.limiter.Wait(ctx); err != nil {
		return fmt.Errorf("context cancelado enquanto aguardava rate limit: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
	if err != nil {
		return err
	}

	resp, err := c.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	return nil
}

---

5. Boas Práticas e Armadilhas de Concorrência

1. Nunca Deixe de Fechar o Canal de Jobs: O loop for job := range jobs nos workers só termina quando o canal é fechado com close(jobs). Se o produtor esquecer de fechar o canal, os workers ficam bloqueados na memória para sempre (Goroutine Leak).
2. Dimensionamento do Pool:
   • Para tarefas intensivas em CPU (criptografia, compressão, processamento de imagens): numWorkers = runtime.NumCPU().
   • Para tarefas intensivas em I/O (consultas de banco, chamadas HTTP, S3): numWorkers = 10 a 100, monitorando o pool de conexões com o banco database-sql e Connection Pooling.
3. Limpeza do Mapa de Rate Limit: Armazenar limitadores por IP em um map sem expiração periódica cria um vazamento de memória contínuo em servidores expostos à internet pública. Use sempre goroutines de expiração com TTL ou armazene contadores no Redis em clusters distribuídos.

---

6. Resumo Conceitual

• Worker Pools protegem o sistema contra concorrência desenfreada mantendo um número fixo de goroutines trabalhadoras.
• Utilize sempre context.Context para que os workers possam ser cancelados graciosamente durante o encerramento do serviço.
• O algoritmo Token Bucket (golang.org/x/time/rate) permite gerenciar taxas médias de transferência ($r$) com suporte a rajadas transitórias ($b$).
• Em APIs HTTP, implemente rate limiting por cliente (IP ou API Key) para evitar que um único atacante degrade o serviço de todos os usuários.
• Em chamadas de saída para APIs de terceiros, use limiter.Wait(ctx) para enfileirar chamadas de forma transparente sem estourar limites de contratos de SLA.