Projeto: worker concorrente

Nesta aula, vamos construir um worker concorrente em Go, um padrão essencial para processamento paralelo de tarefas. Você aprenderá a usar goroutines para execução concorrente, channels para comunicação entre elas, e técnicas para gerenciar o ciclo de vida dos workers com graceful shutdown e context. Este é um projeto prático que consolida conceitos fundamentais de concorrência em Go, preparando você para aplicações reais como processamento de filas, web crawlers e sistemas de eventos.

Dominar esse padrão é crucial para escrever programas eficientes e escaláveis. Vamos começar com os blocos de construção básicos e evoluir até um sistema completo, com foco em boas práticas e tratamento de cancelamento.

Goroutines e channels

Goroutines são funções que executam concorrentemente com outras, mas são leves e gerenciadas pelo runtime do Go. Para criar uma, basta usar a palavra-chave go antes da chamada de função. Channels são a forma principal de comunicação entre goroutines, permitindo enviar e receber valores com segurança. Eles podem ser com buffer ou sem buffer, e o envio/recepção bloqueiam até que o outro lado esteja pronto, o que sincroniza as goroutines automaticamente.

No nosso projeto, usaremos channels para distribuir tarefas e coletar resultados. Vamos criar um canal de tarefas (jobs) e um de resultados (results), onde os workers consomem tarefas e produzem resultados. Essa é a base do padrão worker pool.

Exemplo básico de goroutine e channel:

package main

import (
	"fmt"
	"time"
)

func worker(id int, jobs <-chan int, results chan<- int) {
	for j := range jobs {
		fmt.Printf("worker %d processando job %d\n", id, j)
		time.Sleep(time.Second) // simula trabalho
		results <- j * 2
	}
}

func main() {
	jobs := make(chan int, 10)
	results := make(chan int, 10)

	// inicia 3 workers
	for w := 1; w <= 3; w++ {
		go worker(w, jobs, results)
	}

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

	// coleta resultados
	for r := 1; r <= 5; r++ {
		result := <-results
		fmt.Println("Resultado:", result)
	}
}

Worker pool

Worker pool é um padrão onde um número fixo de goroutines (workers) processa tarefas de uma fila compartilhada. Isso controla o paralelismo e evita sobrecarga de criação de goroutines ilimitadas. No nosso projeto, vamos criar um pool de N workers que consomem tarefas de um canal e enviam resultados para outro. A quantidade de workers pode ser ajustada conforme a carga.

Implementaremos um worker pool com as seguintes características: os workers são iniciados como goroutines, cada um lê do canal de tarefas até que ele seja fechado. O canal de tarefas é fechado quando não há mais tarefas a enviar. Os resultados são enviados para o canal de resultados, que é lido pelo main ou por outra goroutine.

Vamos evoluir nosso exemplo para um worker pool mais completo, com funções específicas:

package main

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

type Task struct {
	ID   int
	Data string
}

type Result struct {
	TaskID int
	Output string
}

func startWorkerPool(numWorkers int, tasks <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 0; i < numWorkers; i++ {
		go worker(i+1, tasks, results)
	}
}

func worker(id int, tasks <-chan Task, results chan<- Result) {
	for task := range tasks {
		// processa a tarefa
		time.Sleep(time.Second)
		result := Result{TaskID: task.ID, Output: fmt.Sprintf("worker %d processou: %s", id, task.Data)}
		results <- result
	}
}

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

	var wg sync.WaitGroup
	wg.Add(1)
	go startWorkerPool(numWorkers, tasks, results, &wg)

	// envia tarefas
	for i := 1; i <= 5; i++ {
		tasks <- Task{ID: i, Data: fmt.Sprintf("tarefa %d", i)}
	}
	close(tasks)

	// coleta resultados
	for i := 1; i <= 5; i++ {
		result := <-results
		fmt.Println("Resultado:", result)
	}

	wg.Wait()
}

Graceful shutdown

Graceful shutdown é o processo de encerrar o programa de forma ordenada, permitindo que as tarefas em andamento sejam concluídas e os recursos sejam liberados corretamente. Em vez de simplesmente matar as goroutines, devemos sinalizar que não há mais trabalho e esperar que elas terminem. Para isso, usamos canais de sinalização e sync.WaitGroup para aguardar a conclusão.

No contexto do worker pool, o graceful shutdown envolve: fechar o canal de tarefas para que os workers terminem seus loops, aguardar que todos os workers terminem (usando WaitGroup), e então fechar o canal de resultados. Também podemos usar um canal de done para sinalizar cancelamento prematuro, se necessário.

Vamos implementar um graceful shutdown com um canal de stop para controlar a parada dos workers:

package main

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

func worker(id int, tasks <-chan int, results chan<- int, stop <-chan struct{}) {
	for {
		select {
		case task, ok := <-tasks:
			if !ok {
				// canal fechado, termina
				return
			}
			fmt.Printf("worker %d processando %d\n", id, task)
			time.Sleep(time.Second)
			results <- task * 2
		case <-stop:
			fmt.Printf("worker %d recebeu stop\n", id)
			return
		}
	}
}

func main() {
	tasks := make(chan int, 10)
	results := make(chan int, 10)
	stop := make(chan struct{})

	var wg sync.WaitGroup
	for w := 1; w <= 3; w++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			worker(id, tasks, results, stop)
		}(w)
	}

	// envia tarefas
	for i := 1; i <= 5; i++ {
		tasks <- i
	}
	close(tasks)

	// simula um sinal de parada após 2 segundos
	time.Sleep(2 * time.Second)
	close(stop)

	// aguarda workers terminarem
	wg.Wait()
	close(results)

	// coleta resultados restantes
	for result := range results {
		fmt.Println("Resultado:", result)
	}
}

Context

O pacote context fornece uma maneira padrão de propagar valores, cancelamentos e deadlines entre goroutines. É especialmente útil para controlar operações de longa duração, como chamadas HTTP ou processamento de tarefas, permitindo cancelamento cooperativo. Com context.Context, podemos sinalizar para os workers que devem parar, com a possibilidade de definir um timeout ou prazo.

No nosso projeto, usaremos context.WithCancel para cancelar os workers quando necessário, substituindo o canal de stop manual. Também podemos usar context.WithTimeout para impor um tempo máximo de execução.

Exemplo de integração do context:

package main

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

func worker(ctx context.Context, id int, tasks <-chan int, results chan<- int) {
	for {
		select {
		case task, ok := <-tasks:
			if !ok {
				return
			}
			fmt.Printf("worker %d processando %d\n", id, task)
			time.Sleep(time.Second)
			results <- task * 2
		case <-ctx.Done():
			fmt.Printf("worker %d cancelado\n", id)
			return
		}
	}
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	tasks := make(chan int, 10)
	results := make(chan int, 10)

	var wg sync.WaitGroup
	for w := 1; w <= 3; w++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			worker(ctx, id, tasks, results)
		}(w)
	}

	// envia tarefas
	for i := 1; i <= 5; i++ {
		tasks <- i
	}
	close(tasks)

	// cancela após 2 segundos (simulando um timeout)
	time.Sleep(2 * time.Second)
	cancel()

	wg.Wait()
	close(results)

	for result := range results {
		fmt.Println("Resultado:", result)
	}
}

Boas práticas e observações finais

Ao construir workers concorrentes, é importante seguir algumas boas práticas: sempre fechar canais quando não houver mais produtores para evitar deadlocks; usar sync.WaitGroup para aguardar a conclusão das goroutines; usar select com canais para lidar com múltiplas fontes de eventos; e usar context para cancelamento e timeouts. Evite compartilhar estado entre goroutines sem sincronização; prefira comunicação via channels.

Também é útil definir um limite para o número de workers para evitar sobrecarga de recursos. Teste seu código com casos de borda, como canal fechado inesperadamente ou cancelamento prematuro. Lembre-se de que a concorrência em Go é poderosa, mas exige cuidado com a sincronização.

Referências

• A Tour of Go - Goroutines
• A Tour of Go - Channels
• Effective Go - Concurrency
• Documentação do pacote context
• Documentação do sync.WaitGroup
• Go Blog: Pipelines and cancellation
• Go Blog: Context and Cancellation

Exercícios

1. Crie um worker pool com 4 workers que processam uma lista de 10 números, calculando o quadrado de cada um. Imprima os resultados.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"fmt"
   	"sync"
   )
   
   func worker(id int, jobs <-chan int, results chan<- int) {
   	for j := range jobs {
   		results <- j * j
   	}
   }
   
   func main() {
   	const numJobs = 10
   	jobs := make(chan int, numJobs)
   	results := make(chan int, numJobs)
   
   	var wg sync.WaitGroup
   	for w := 1; w <= 4; w++ {
   		wg.Add(1)
   		go func(id int) {
   			defer wg.Done()
   			worker(id, jobs, results)
   		}(w)
   	}
   
   	for j := 1; j <= numJobs; j++ {
   		jobs <- j
   	}
   	close(jobs)
   
   	wg.Wait()
   	close(results)
   
   	for r := range results {
   		fmt.Println(r)
   	}
   }
   

2. Modifique o worker pool para usar context.WithTimeout que cancela após 2 segundos. Envie 10 tarefas que dormem 500ms cada. Observe que nem todas serão processadas.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"context"
   	"fmt"
   	"sync"
   	"time"
   )
   
   func worker(ctx context.Context, id int, jobs <-chan int, results chan<- int) {
   	for {
   		select {
   		case j, ok := <-jobs:
   			if !ok {
   				return
   			}
   			time.Sleep(500 * time.Millisecond)
   			results <- j * j
   		case <-ctx.Done():
   			return
   		}
   	}
   }
   
   func main() {
   	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
   	defer cancel()
   
   	jobs := make(chan int, 10)
   	results := make(chan int, 10)
   
   	var wg sync.WaitGroup
   	for w := 1; w <= 3; w++ {
   		wg.Add(1)
   		go func(id int) {
   			defer wg.Done()
   			worker(ctx, id, jobs, results)
   		}(w)
   	}
   
   	for j := 1; j <= 10; j++ {
   		jobs <- j
   	}
   	close(jobs)
   
   	wg.Wait()
   	close(results)
   
   	for r := range results {
   		fmt.Println(r)
   	}
   }
   

3. Implemente um graceful shutdown onde o programa aguarda todas as tarefas terminarem antes de sair, mesmo que um sinal de interrupção (Ctrl+C) seja recebido. Use signal.Notify.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"context"
   	"fmt"
   	"os"
   	"os/signal"
   	"sync"
   	"syscall"
   	"time"
   )
   
   func worker(ctx context.Context, id int, jobs <-chan int, results chan<- int) {
   	for {
   		select {
   		case j, ok := <-jobs:
   			if !ok {
   				return
   			}
   			time.Sleep(time.Second)
   			results <- j * 2
   		case <-ctx.Done():
   			return
   		}
   	}
   }
   
   func main() {
   	ctx, cancel := context.WithCancel(context.Background())
   	defer cancel()
   
   	jobs := make(chan int, 10)
   	results := make(chan int, 10)
   
   	var wg sync.WaitGroup
   	for w := 1; w <= 3; w++ {
   		wg.Add(1)
   		go func(id int) {
   			defer wg.Done()
   			worker(ctx, id, jobs, results)
   		}(w)
   	}
   
   	// enviar tarefas em uma goroutine para não bloquear
   	go func() {
   		for i := 1; i <= 5; i++ {
   			jobs <- i
   		}
   		close(jobs)
   	}()
   
   	// captura sinal de interrupção
   	sigChan := make(chan os.Signal, 1)
   	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
   
   	go func() {
   		<-sigChan
   		fmt.Println("Sinal recebido, cancelando...")
   		cancel()
   	}()
   
   	// aguarda workers terminarem
   	wg.Wait()
   	close(results)
   
   	for r := range results {
   		fmt.Println("Resultado:", r)
   	}
   }
   

4. Crie um worker pool onde cada worker tem um ID e processa tarefas de um canal, mas com um limite de tempo por tarefa. Use context.WithTimeout por tarefa.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"context"
   	"fmt"
   	"sync"
   	"time"
   )
   
   func processTask(ctx context.Context, task int) (int, error) {
   	select {
   	case <-time.After(200 * time.Millisecond):
   		return task * task, nil
   	case <-ctx.Done():
   		return 0, ctx.Err()
   	}
   }
   
   func worker(id int, jobs <-chan int, results chan<- int) {
   	for task := range jobs {
   		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
   		res, err := processTask(ctx, task)
   		cancel()
   		if err != nil {
   			fmt.Printf("worker %d falhou na tarefa %d: %v\n", id, task, err)
   			continue
   		}
   		results <- res
   	}
   }
   
   func main() {
   	jobs := make(chan int, 5)
   	results := make(chan int, 5)
   
   	var wg sync.WaitGroup
   	for w := 1; w <= 2; w++ {
   		wg.Add(1)
   		go func(id int) {
   			defer wg.Done()
   			worker(id, jobs, results)
   		}(w)
   	}
   
   	for i := 1; i <= 5; i++ {
   		jobs <- i
   	}
   	close(jobs)
   
   	wg.Wait()
   	close(results)
   
   	for r := range results {
   		fmt.Println("Resultado:", r)
   	}
   }
   

5. Implemente um worker pool que processa uma lista de URLs, baixando o conteúdo (simulado) e contando palavras. Use context para cancelar se demorar mais que 3 segundos no total.

   Ver resposta

   ✓ Resposta:

   package main
   
   import (
   	"context"
   	"fmt"
   	"strings"
   	"sync"
   	"time"
   )
   
   func fetchURL(ctx context.Context, url string) (string, error) {
   	select {
   	case <-time.After(500 * time.Millisecond):
   		// simula conteúdo
   		return "conteúdo da página " + url, nil
   	case <-ctx.Done():
   		return "", ctx.Err()
   	}
   }
   
   func countWords(content string) int {
   	return len(strings.Fields(content))
   }
   
   func main() {
   	urls := []string{"http://exemplo.com/1", "http://exemplo.com/2", "http://exemplo.com/3", "http://exemplo.com/4", "http://exemplo.com/5"}
   	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
   	defer cancel()
   
   	jobs := make(chan string, len(urls))
   	results := make(chan int, len(urls))
   
   	var wg sync.WaitGroup
   	for w := 1; w <= 3; w++ {
   		wg.Add(1)
   		go func() {
   			defer wg.Done()
   			for {
   				select {
   				case url, ok := <-jobs:
   					if !ok {
   						return
   					}
   					content, err := fetchURL(ctx, url)
   					if err != nil {
   						fmt.Println("Erro ao buscar", url, ":", err)
   						continue
   					}
   					results <- countWords(content)
   				case <-ctx.Done():
   					return
   				}
   			}
   		}()
   	}
   
   	for _, url := range urls {
   		jobs <- url
   	}
   	close(jobs)
   
   	wg.Wait()
   	close(results)
   
   	total := 0
   	for count := range results {
   		total += count
   	}
   	fmt.Println("Total de palavras:", total)
   }