Padrões de concorrência

Padrões de concorrência são soluções reutilizáveis para problemas comuns em programas concorrentes. Em Go, a combinação de goroutines e canais permite implementar esses padrões de forma elegante e segura. Nesta aula, exploraremos quatro padrões fundamentais: Worker pool, Fan-in/Fan-out, Pipeline e Rate limiting. Cada um resolve um desafio específico de concorrência, como controlar o número de tarefas simultâneas, distribuir trabalho, processar dados em estágios e limitar a taxa de operações.

Entender esses padrões é essencial para escrever programas Go que aproveitem ao máximo o paralelismo, evitando problemas como contenção de recursos, deadlocks e sobrecarga. Vamos mergulhar em cada padrão com exemplos práticos.

Worker pool

O padrão Worker pool mantém um número fixo de goroutines (workers) que processam tarefas de uma fila compartilhada. É útil quando você precisa controlar a concorrência para evitar sobrecarregar recursos como conexões de rede ou uso de CPU. Em Go, implementamos usando um canal de tarefas e um canal de resultados.

Os workers consomem tarefas do canal de entrada e enviam resultados para o canal de saída. O número de workers determina o grau de concorrência. Esse padrão é comum em servidores web, processamento de lotes e sistemas de filas.

package main

import (
	"fmt"
	"time"
)

func worker(id int, tasks <-chan int, results chan<- int) {
	for task := range tasks {
		fmt.Printf("Worker %d processing task %d\n", id, task)
		time.Sleep(time.Second)
		results <- task * 2
	}
}

func main() {
	numWorkers := 3
	tasks := make(chan int, 10)
	results := make(chan int, 10)

	// Start workers
	for i := 1; i <= numWorkers; i++ {
		go worker(i, tasks, results)
	}

	// Send tasks
	for j := 1; j <= 5; j++ {
		tasks <- j
	}
	close(tasks)

	// Collect results
	for k := 1; k <= 5; k++ {
		result := <-results
		fmt.Printf("Result: %d\n", result)
	}
}

Neste exemplo, três workers processam cinco tarefas. O canal tasks é fechado após enviar todas as tarefas, fazendo com que os workers saiam do loop ao consumir todas. Os resultados são coletados em ordem de conclusão, que pode ser diferente da ordem de envio.

Fan-in/fan-out

Fan-out é quando uma única fonte de dados é distribuída para múltiplas goroutines para processamento paralelo. Fan-in é a coleta dos resultados de múltiplas goroutines em um único canal. Esses padrões são frequentemente usados juntos: fan-out para paralelizar o trabalho e fan-in para agregar os resultados.

Em Go, fan-out pode ser feito enviando o mesmo canal para várias goroutines. Fan-in geralmente usa uma goroutine dedicada que lê de vários canais de entrada e combina as saídas em um único canal, ou simplesmente coleta os resultados diretamente no main.

package main

import (
	"fmt"
	"sync"
)

func producer(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}

func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			out <- n * n
		}
	}()
	return out
}

func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	output := func(c <-chan int) {
		defer wg.Done()
		for n := range c {
			out <- n
		}
	}

	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

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

func main() {
	in := producer(1, 2, 3, 4)

	// Fan-out: distribuir para dois workers
	c1 := square(in)
	c2 := square(in)

	// Fan-in: combinar resultados
	for result := range merge(c1, c2) {
		fmt.Println(result)
	}
}

Aqui, producer gera números, square processa cada número (fan-out), e merge combina os canais de saída (fan-in). O sync.WaitGroup garante que todos os canais sejam lidos antes de fechar o canal de saída.

Pipeline

Pipeline é uma série de estágios conectados por canais, onde cada estágio processa dados e os passa para o próximo. Cada estágio pode ser executado por uma ou mais goroutines, permitindo paralelismo. Pipelines são ideais para processamento de dados em fluxo, como transformações de dados, ETL e processamento de logs.

Em Go, um pipeline típico tem um estágio de geração, um ou mais estágios de processamento e um estágio de consumo. Canais são usados para conectar os estágios. Fechar canais corretamente é essencial para evitar deadlocks.

package main

import "fmt"

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			out <- n * n
		}
	}()
	return out
}

func main() {
	// Set up the pipeline
	c := gen(2, 3, 4)
	out := sq(c)

	// Consume the output
	for result := range out {
		fmt.Println(result)
	}
}

Este pipeline simples tem dois estágios: gen gera números e sq calcula o quadrado. Cada estágio retorna um canal de saída. O consumo é feito no main. Para pipelines mais complexos, você pode adicionar mais estágios e paralelizar cada estágio com múltiplas goroutines.

Rate limiting

Rate limiting controla a frequência com que operações são executadas, evitando sobrecarga em sistemas externos (APIs, bancos de dados) ou no próprio sistema. Em Go, podemos implementar rate limiting com time.Ticker ou usando um canal com buffer e goroutines.

O padrão mais comum é usar um ticker que libera tokens em intervalos regulares. Cada operação consome um token antes de prosseguir. Isso garante que as operações ocorram dentro de um limite máximo por segundo.

package main

import (
	"fmt"
	"time"
)

func main() {
	requests := make(chan int, 5)
	for i := 1; i <= 5; i++ {
		requests <- i
	}
	close(requests)

	limiter := time.Tick(200 * time.Millisecond)

	for req := range requests {
		<-limiter // wait for token
		fmt.Println("Processing request", req, "at", time.Now())
	}
}

Aqui, time.Tick cria um canal que envia um valor a cada 200ms. O loop consome um token antes de processar cada requisição, limitando a taxa a 5 requisições por segundo. Para permitir rajadas, podemos usar um canal com buffer (bucket de tokens).

package main

import (
	"fmt"
	"time"
)

func main() {
	burstyLimiter := make(chan time.Time, 3)

	// Preencher o canal com tokens iniciais (rajada)
	for i := 0; i < 3; i++ {
		burstyLimiter <- time.Now()
	}

	// Recarregar tokens a cada 200ms
	go func() {
		for t := range time.Tick(200 * time.Millisecond) {
			burstyLimiter <- t
		}
	}()

	burstyRequests := make(chan int, 5)
	for i := 1; i <= 5; i++ {
		burstyRequests <- i
	}
	close(burstyRequests)

	for req := range burstyRequests {
		<-burstyLimiter
		fmt.Println("Processing request", req, "at", time.Now())
	}
}

Neste exemplo, o limitador permite uma rajada inicial de até 3 requisições imediatas, depois limita a 5 por segundo (200ms entre tokens).

Boas práticas

Ao usar padrões de concorrência em Go, lembre-se: sempre feche canais quando não houver mais dados a enviar para evitar deadlocks; use sync.WaitGroup para sincronizar goroutines; evite compartilhar memória, prefira comunicar via canais; e teste com a flag -race para detectar condições de corrida.

Referências

• Go Blog: Pipelines
• Effective Go
• Go Language Specification
• Go by Example: Worker Pools
• Go by Example: Rate Limiting
• Package time (Go docs)
• Package sync (Go docs)

Exercícios

1. Implemente um worker pool que processa URLs de uma lista, fazendo uma requisição HTTP para cada uma e imprimindo o status code. Use 4 workers.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"fmt"
   	"net/http"
   	"sync"
   )
   
   func worker(id int, urls <-chan string, wg *sync.WaitGroup) {
   	defer wg.Done()
   	for url := range urls {
   		resp, err := http.Get(url)
   		if err != nil {
   			fmt.Printf("Worker %d: error fetching %s: %v\n", id, url, err)
   			continue
   		}
   		fmt.Printf("Worker %d: %s - %d\n", id, url, resp.StatusCode)
   		resp.Body.Close()
   	}
   }
   
   func main() {
   	urls := []string{"https://go.dev", "https://golang.org", "https://example.com", "https://httpbin.org/status/200", "https://httpbin.org/status/404"}
   	numWorkers := 4
   	urlChan := make(chan string, len(urls))
   	var wg sync.WaitGroup
   
   	for i := 1; i <= numWorkers; i++ {
   		wg.Add(1)
   		go worker(i, urlChan, &wg)
   	}
   
   	for _, url := range urls {
   		urlChan <- url
   	}
   	close(urlChan)
   
   	wg.Wait()
   }
   

2. Crie um pipeline de três estágios: o primeiro gera números de 1 a 10, o segundo eleva ao cubo, e o terceiro imprime o resultado. Use canais.

   Ver resposta

   ✓ Resposta:

   package main
   
   import "fmt"
   
   func generate(nums ...int) <-chan int {
   	out := make(chan int)
   	go func() {
   		defer close(out)
   		for _, n := range nums {
   			out <- n
   		}
   	}()
   	return out
   }
   
   func cube(in <-chan int) <-chan int {
   	out := make(chan int)
   	go func() {
   		defer close(out)
   		for n := range in {
   			out <- n * n * n
   		}
   	}()
   	return out
   }
   
   func printResults(in <-chan int) {
   	for n := range in {
   		fmt.Println(n)
   	}
   }
   
   func main() {
   	numbers := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
   	cubed := cube(numbers)
   	printResults(cubed)
   }
   

3. Implemente fan-out/fan-in para calcular o fatorial de vários números. Use 3 workers para calcular fatoriais e um merge para coletar resultados.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"fmt"
   	"sync"
   )
   
   func factorial(n int) int {
   	if n <= 1 {
   		return 1
   	}
   	result := 1
   	for i := 2; i <= n; i++ {
   		result *= i
   	}
   	return result
   }
   
   func producer(nums ...int) <-chan int {
   	out := make(chan int)
   	go func() {
   		defer close(out)
   		for _, n := range nums {
   			out <- n
   		}
   	}()
   	return out
   }
   
   func worker(in <-chan int) <-chan struct{ num, fact int } {
   	out := make(chan struct{ num, fact int })
   	go func() {
   		defer close(out)
   		for n := range in {
   			out <- struct{ num, fact int }{n, factorial(n)}
   		}
   	}()
   	return out
   }
   
   func merge(cs ...<-chan struct{ num, fact int }) <-chan struct{ num, fact int } {
   	var wg sync.WaitGroup
   	out := make(chan struct{ num, fact int })
   
   	output := func(c <-chan struct{ num, fact int }) {
   		defer wg.Done()
   		for v := range c {
   			out <- v
   		}
   	}
   
   	wg.Add(len(cs))
   	for _, c := range cs {
   		go output(c)
   	}
   
   	go func() {
   		wg.Wait()
   		close(out)
   	}()
   	return out
   }
   
   func main() {
   	nums := []int{5, 7, 10, 3, 6}
   	in := producer(nums...)
   
   	// Fan-out: 3 workers
   	c1 := worker(in)
   	c2 := worker(in)
   	c3 := worker(in)
   
   	// Fan-in
   	for result := range merge(c1, c2, c3) {
   		fmt.Printf("Factorial of %d is %d\n", result.num, result.fact)
   	}
   }
   

4. Crie um rate limiter que permita no máximo 2 requisições por segundo, com rajada de 1 requisição inicial. Teste com 6 requisições.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"fmt"
   	"time"
   )
   
   func main() {
   	requests := make(chan int, 6)
   	for i := 1; i <= 6; i++ {
   		requests <- i
   	}
   	close(requests)
   
   	limiter := make(chan time.Time, 1)
   	// Rajada inicial: um token
   	limiter <- time.Now()
   
   	go func() {
   		for t := range time.Tick(500 * time.Millisecond) {
   			limiter <- t
   		}
   	}()
   
   	for req := range requests {
   		<-limiter
   		fmt.Println("Processing request", req, "at", time.Now())
   	}
   }
   

5. Modifique o worker pool do exercício 1 para usar um contexto com timeout de 2 segundos. Se uma requisição demorar mais, cancele e registre o erro.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"context"
   	"fmt"
   	"net/http"
   	"sync"
   	"time"
   )
   
   func worker(ctx context.Context, id int, urls <-chan string, wg *sync.WaitGroup) {
   	defer wg.Done()
   	for url := range urls {
   		req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
   		if err != nil {
   			fmt.Printf("Worker %d: error creating request for %s: %v\n", id, url, err)
   			continue
   		}
   		resp, err := http.DefaultClient.Do(req)
   		if err != nil {
   			fmt.Printf("Worker %d: error fetching %s: %v\n", id, url, err)
   			continue
   		}
   		fmt.Printf("Worker %d: %s - %d\n", id, url, resp.StatusCode)
   		resp.Body.Close()
   	}
   }
   
   func main() {
   	urls := []string{"https://go.dev", "https://golang.org", "https://example.com", "https://httpbin.org/delay/3", "https://httpbin.org/status/200"}
   	numWorkers := 4
   	urlChan := make(chan string, len(urls))
   	var wg sync.WaitGroup
   
   	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
   	defer cancel()
   
   	for i := 1; i <= numWorkers; i++ {
   		wg.Add(1)
   		go worker(ctx, i, urlChan, &wg)
   	}
   
   	for _, url := range urls {
   		urlChan <- url
   	}
   	close(urlChan)
   
   	wg.Wait()
   }