Concorrência em cenários reais

Esta aula explora padrões avançados de concorrência em Go aplicados a cenários reais, incluindo coordenação de tarefas, timeouts e cancelamento, backpressure e padrões de design. Você aprenderá a usar canais, sync.WaitGroup, context, e técnicas como worker pools e pipeline para construir sistemas concorrentes robustos e eficientes.

Nesta aula, vamos mergulhar em padrões de concorrência que são essenciais para construir sistemas reais em Go. Você já conhece os fundamentos — goroutines, canais e sincronização — mas agora vamos aplicá-los em situações que você encontrará no dia a dia: coordenar múltiplas tarefas, respeitar prazos, controlar a carga de trabalho e lidar com falhas. Esses padrões não apenas tornam seu código mais eficiente, mas também mais previsível e resiliente.

Vamos explorar desde a coordenação básica com sync.WaitGroup até padrões mais sofisticados como worker pools e pipeline, sempre com exemplos práticos e explicações detalhadas. Ao final, você terá um repertório de técnicas para resolver problemas de concorrência com confiança.

Coordenação de tarefas

Em programas concorrentes, frequentemente precisamos lançar várias goroutines e esperar que todas terminem antes de prosseguir. A forma mais simples é usar sync.WaitGroup, que permite contar quantas goroutines estão em execução e bloquear até que todas sinalizem sua conclusão.

O WaitGroup possui três métodos principais: Add, Done e Wait. Você chama Add antes de iniciar cada goroutine, Done quando a goroutine termina, e Wait para bloquear até que o contador chegue a zero. É crucial que Add seja chamado antes da goroutine ser lançada, para evitar corridas.

package main

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

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("Worker %d iniciando\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d concluído\n", id)
}

func main() {
    var wg sync.WaitGroup
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }
    wg.Wait()
    fmt.Println("Todos os workers concluídos")
}

Em cenários mais complexos, você pode combinar WaitGroup com canais para coletar resultados. Por exemplo, imagine que você precisa processar uma lista de URLs e coletar os status HTTP. Você pode lançar uma goroutine para cada URL e enviar os resultados para um canal, enquanto outra goroutine fecha o canal quando todas terminarem.

func main() {
    urls := []string{"http://example.com", "http://golang.org", "http://github.com"}
    results := make(chan string)
    var wg sync.WaitGroup

    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            resp, err := http.Get(u)
            if err != nil {
                results <- fmt.Sprintf("%s: erro: %v", u, err)
                return
            }
            results <- fmt.Sprintf("%s: %s", u, resp.Status)
            resp.Body.Close()
        }(url)
    }

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

    for res := range results {
        fmt.Println(res)
    }
}

Timeouts e cancelamento

Trabalhar com operações que podem demorar indefinidamente (como chamadas de rede ou leitura de arquivos) exige controle de tempo e cancelamento. O pacote context é a ferramenta padrão para isso. Ele permite definir prazos (deadlines) ou cancelar operações manualmente, e propagar essa informação para goroutines filhas.

Um uso comum é criar um contexto com timeout e passá-lo para funções que fazem requisições HTTP. Se o tempo expirar, a requisição é cancelada automaticamente. Você também pode usar context.WithCancel para cancelar manualmente quando uma condição for satisfeita (por exemplo, quando um usuário interrompe uma busca).

func fetchWithTimeout(ctx context.Context, url string) error {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return err
    }
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    // processar resposta
    return nil
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    if err := fetchWithTimeout(ctx, "http://example.com"); err != nil {
        fmt.Println("Erro:", err)
    }
}

Para cancelamento manual, você pode usar context.WithCancel e chamar cancel() quando necessário. Por exemplo, em um servidor que processa solicitações, você pode cancelar o contexto quando o cliente desconectar. As goroutines que recebem esse contexto devem verificar ctx.Done() periodicamente para sair de forma limpa.

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    go func() {
        time.Sleep(2 * time.Second)
        cancel() // cancela após 2 segundos
    }()

    select {
    case <-ctx.Done():
        fmt.Println("Contexto cancelado:", ctx.Err())
    }
}

Backpressure

Backpressure é um mecanismo para controlar o fluxo de dados entre produtores e consumidores, evitando que um produtor rápido sobrecarregue um consumidor lento. Em Go, os canais com buffer limitado podem servir como uma forma de backpressure: se o buffer estiver cheio, o produtor bloqueia até que o consumidor libere espaço.

No entanto, em sistemas mais complexos, você pode precisar de um controle mais fino, como limitar o número de requisições simultâneas ou usar um token bucket. O padrão de worker pool (que veremos em Padrões) também ajuda a aplicar backpressure limitando o número de goroutines que processam trabalho.

Um exemplo simples de backpressure com canal bufferizado:

func main() {
    jobs := make(chan int, 3) // buffer de tamanho 3

    // produtor
    go func() {
        for i := 0; i < 10; i++ {
            jobs <- i
            fmt.Println("Enviado:", i)
        }
        close(jobs)
    }()

    // consumidor
    for job := range jobs {
        fmt.Println("Processando:", job)
        time.Sleep(100 * time.Millisecond)
    }
}

Nesse exemplo, o produtor só consegue enviar quando o consumidor lê, pois o buffer tem tamanho 3. Se o consumidor for lento, o produtor bloqueia, aplicando backpressure natural.

Outra técnica é usar um semáforo com canais para limitar a concorrência. Você cria um canal com buffer de tamanho N e cada goroutine adquire um token antes de executar, liberando-o após terminar. Isso limita o número de operações simultâneas.

var sem = make(chan struct{}, 3) // permite 3 concorrentes

func process(item int) {
    sem <- struct{}{} // adquire token
    defer func() { <-sem }() // libera token
    fmt.Println("Processando", item)
    time.Sleep(time.Second)
}

func main() {
    for i := 0; i < 10; i++ {
        go process(i)
    }
    time.Sleep(5 * time.Second) // espera todos
}

Padrões

Existem vários padrões de concorrência consagrados em Go. Vamos destacar três: Worker Pool, Pipeline e Fan-out/Fan-in.

Worker Pool é um conjunto fixo de goroutines que processam tarefas de uma fila. Ele controla a concorrência e usa canais para distribuir trabalho. É ideal para tarefas CPU-bound ou I/O-bound, como processamento de imagens ou envio de e-mails.

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)
        results <- j * 2
    }
}

func main() {
    const numJobs = 10
    const numWorkers = 3
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)

    for w := 1; w <= numWorkers; w++ {
        go worker(w, jobs, results)
    }

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

    for r := 1; r <= numJobs; r++ {
        <-results
    }
}

Pipeline é uma série de estágios conectados por canais, onde cada estágio processa dados e os envia para o próximo. É útil para processamento em etapas, como transformar dados, filtrar, agregar. Cada estágio pode ser executado em uma goroutine separada, e os canais fazem a comunicação.

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

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

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

    // Consume the output.
    for v := range out {
        fmt.Println(v) // 4, 9, 16
    }
}

Fan-out/Fan-in envolve distribuir tarefas para várias goroutines (fan-out) e depois combinar os resultados em um único canal (fan-in). É uma generalização do worker pool, mas com canais de entrada e saída.

func worker(id int, in <-chan int, out chan<- int) {
    for v := range in {
        out <- v * 2
    }
}

func main() {
    input := make(chan int)
    output := make(chan int)

    // Lança 3 workers
    for i := 0; i < 3; i++ {
        go worker(i, input, output)
    }

    // Alimenta o input
    go func() {
        for i := 1; i <= 10; i++ {
            input <- i
        }
        close(input)
    }()

    // Coleta os resultados
    for i := 0; i < 10; i++ {
        result := <-output
        fmt.Println(result)
    }
}

Esses padrões são a base para construir sistemas concorrentes robustos. Cada um tem seus trade-offs: worker pool controla a concorrência, pipeline é ótimo para streaming, e fan-out/fan-in maximiza o paralelismo.

Boas práticas

Ao trabalhar com concorrência em Go, algumas boas práticas podem evitar bugs e melhorar a legibilidade:

Não compartilhe memória; comunique-se através de canais.
Evite goroutines vazando: sempre tenha um mecanismo de cancelamento (context) ou feche canais adequadamente.
Use sync.WaitGroup para sincronizar a conclusão de goroutines, mas não o utilize para passar dados.
Prefira canais com buffer pequeno ou unbuffered para comunicação síncrona, a menos que precise de backpressure.
Teste seu código com a flag -race para detectar corridas de dados.
Documente o fluxo de dados entre goroutines para facilitar a manutenção.

Referências

Exercícios

Escreva um programa que lance 10 goroutines, cada uma imprimindo seu ID, e use sync.WaitGroup para garantir que o programa espere todas terminarem antes de imprimir "Fim".

✓ Resposta:

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Println("Goroutine", id)
        }(i)
    }
    wg.Wait()
    fmt.Println("Fim")
}

Implemente uma função que receba um contexto e um canal de inteiros, e que leia valores do canal até que o contexto seja cancelado ou o canal seja fechado. Ao final, imprima quantos valores foram lidos.

✓ Resposta:

func readUntilCancel(ctx context.Context, ch <-chan int) int {
    count := 0
    for {
        select {
        case <-ctx.Done():
            return count
        case v, ok := <-ch:
            if !ok {
                return count
            }
            fmt.Println("Valor:", v)
            count++
        }
    }
}

Crie um worker pool com 4 workers que processam 20 tarefas. Cada tarefa é um número; o worker deve calcular o quadrado e enviar o resultado para um canal de resultados. Imprima todos os resultados.

✓ Resposta:

package main

import (
    "fmt"
    "sync"
)

func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        results <- j * j
    }
}

func main() {
    const numJobs = 20
    const numWorkers = 4
    jobs := make(chan int, numJobs)
    results := make(chan int, numJobs)
    var wg sync.WaitGroup

    for w := 0; w < numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

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

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

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

Implemente um pipeline com três estágios: geração de números de 1 a 10, multiplicação por 2 e filtro de números pares (ou seja, manter apenas os pares). Use canais e goroutines.

✓ Resposta:

package main

import "fmt"

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

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

func filterEven(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            if n%2 == 0 {
                out <- n
            }
        }
        close(out)
    }()
    return out
}

func main() {
    c := gen(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    c = multiply(c, 2)
    c = filterEven(c)
    for v := range c {
        fmt.Println(v)
    }
}

Escreva um programa que simule backpressure: um produtor gera números de 1 a 100, mas o consumidor só pode processar 5 por segundo. Use um canal bufferizado de tamanho 5 e um time.Sleep no consumidor. Observe o comportamento e explique em comentários.

✓ Resposta:

package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan int, 5) // buffer de 5

    // Produtor
    go func() {
        for i := 1; i <= 100; i++ {
            ch <- i
            fmt.Println("Produziu", i)
        }
        close(ch)
    }()

    // Consumidor
    for v := range ch {
        fmt.Println("Consumiu", v)
        time.Sleep(200 * time.Millisecond) // 5 por segundo
    }
}

Concorrência em cenários reais

Coordenação de tarefas

Timeouts e cancelamento

Backpressure

Padrões

Boas práticas

Referências

Exercícios

Continue lendo

Observabilidade

Escrevendo Go idiomático

Genéricos em cenários reais