Bem-vindo à aula 90 do nosso curso de Go! Hoje vamos colocar em prática vários conceitos importantes que estudamos ao longo do curso, construindo um projeto completo: um coletor concorrente de dados. Esse tipo de ferramenta é comum em sistemas que precisam buscar informações de várias fontes ao mesmo tempo, como monitores de preços, agregadores de notícias ou verificadores de disponibilidade de serviços.

O objetivo é criar um programa que faça requisições HTTP para uma lista de URLs, processe as respostas de forma concorrente, respeite um limite de requisições por segundo e, ao final, agregue os resultados em um relatório. Também vamos lidar com erros de forma elegante, garantindo que uma falha em uma requisição não derrube todo o processo. Vamos mergulhar nos detalhes.

Requisições HTTP concorrentes

O coração do nosso coletor é a capacidade de fazer muitas requisições HTTP ao mesmo tempo. Em Go, a concorrência é nativa e podemos usar goroutines para disparar várias requisições simultaneamente. No entanto, precisamos ter cuidado para não sobrecarregar o servidor ou o próprio programa com um número excessivo de goroutines. Por isso, usaremos um padrão de worker pool com canais.

Primeiro, vamos definir a estrutura básica: uma função que recebe uma URL e retorna o status code e o tamanho do corpo. Em seguida, criaremos um pool de workers, onde cada worker consome URLs de um canal e envia os resultados para outro canal. O número de workers controla o grau de concorrência.

package main

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

type Result struct {
    URL        string
    StatusCode int
    Size       int
    Err        error
}

func fetch(url string) Result {
    resp, err := http.Get(url)
    if err != nil {
        return Result{URL: url, Err: err}
    }
    defer resp.Body.Close()

    // Lê o corpo para obter o tamanho (simplificado)
    buf := make([]byte, 1024)
    size := 0
    for {
        n, err := resp.Body.Read(buf)
        size += n
        if err != nil {
            break
        }
    }
    return Result{URL: url, StatusCode: resp.StatusCode, Size: size}
}

func worker(id int, urls <-chan string, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for url := range urls {
        fmt.Printf("Worker %d: buscando %s\n", id, url)
        results <- fetch(url)
    }
}

func main() {
    urls := []string{
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/404",
        "https://httpbin.org/status/500",
        "https://httpbin.org/html",
        "https://httpbin.org/robots.txt",
    }

    numWorkers := 3
    urlsChan := make(chan string)
    resultsChan := make(chan Result)
    var wg sync.WaitGroup

    // Inicia workers
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go worker(i, urlsChan, resultsChan, &wg)
    }

    // Envia URLs para o canal
    go func() {
        defer close(urlsChan)
        for _, url := range urls {
            urlsChan <- url
        }
    }()

    // Coleta resultados em uma goroutine separada
    go func() {
        wg.Wait()
        close(resultsChan)
    }()

    // Processa resultados
    for res := range resultsChan {
        if res.Err != nil {
            fmt.Printf("Erro em %s: %v\n", res.URL, res.Err)
            continue
        }
        fmt.Printf("%s - Status: %d - Size: %d bytes\n", res.URL, res.StatusCode, res.Size)
    }
}

No exemplo acima, usamos três workers para processar cinco URLs. O canal urlsChan distribui as tarefas, e o canal resultsChan coleta os resultados. O sync.WaitGroup garante que não fecharemos o canal de resultados antes que todos os workers terminem.

É importante notar que a ordem de saída não é determinística, pois as goroutines executam de forma independente. Se precisar preservar a ordem original, você pode associar um índice a cada URL e ordenar os resultados depois.

Limitação de taxa

Quando fazemos muitas requisições concorrentes, podemos facilmente ultrapassar limites de API ou sobrecarregar um servidor. A limitação de taxa (rate limiting) é essencial para controlar a frequência de requisições. Em Go, podemos implementar isso com um time.Ticker ou usando a biblioteca golang.org/x/time/rate. Vamos usar a abordagem com time.Ticker, que é simples e eficaz.

A ideia é criar um ticker que emite um sinal a cada intervalo definido. Antes de cada requisição, o worker espera um sinal do ticker. Isso garante que não faremos mais do que um certo número de requisições por segundo.

// Exemplo de limitação de taxa com ticker
func rateLimitedFetch(url string, limiter *time.Ticker) Result {
    <-limiter.C  // espera o próximo tick
    return fetch(url)
}

func workerRateLimited(id int, urls <-chan string, results chan<- Result, limiter *time.Ticker, wg *sync.WaitGroup) {
    defer wg.Done()
    for url := range urls {
        results <- rateLimitedFetch(url, limiter)
    }
}

No main, criaríamos um ticker com o intervalo desejado. Por exemplo, para permitir 2 requisições por segundo, usamos time.NewTicker(500 * time.Millisecond).

Outra abordagem mais robusta é usar o pacote golang.org/x/time/rate, que oferece um limitador com suporte a rajadas (bursts). Isso é útil quando você quer permitir um pequeno pico inicial. Vamos ver um exemplo:

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

func main() {
    limiter := rate.NewLimiter(rate.Limit(2), 5) // 2 req/s, burst de 5
    // ...
    if limiter.Allow() {
        // faz a requisição
    } else {
        // espera ou rejeita
    }
}

O método Allow() retorna true se a requisição pode ser feita agora, e false caso contrário. Você pode combinar isso com um loop de espera para não perder requisições. No nosso projeto, usaremos o ticker para simplificar, mas mencionamos o pacote rate como alternativa profissional.

Agregação

Depois de coletar os resultados, precisamos agregá-los de forma útil. Nossa agregação pode ser simples: contar quantas respostas foram bem-sucedidas, quantas falharam, calcular o tempo total e talvez gerar estatísticas sobre os códigos de status. Vamos estender nosso programa para exibir um relatório final.

Para isso, podemos usar um mapa para armazenar contagens por código de status e um acumulador para o tamanho total. Também podemos medir o tempo de execução com time.Now() e time.Since().

func aggregate(results <-chan Result) map[int]int {
    statusCount := make(map[int]int)
    for res := range results {
        if res.Err != nil {
            statusCount[-1]++ // código especial para erros
            continue
        }
        statusCount[res.StatusCode]++
    }
    return statusCount
}

func main() {
    start := time.Now()
    // ... (configuração e execução)

    statusCount := aggregate(resultsChan)

    fmt.Println("\nRelatório:")
    for status, count := range statusCount {
        if status == -1 {
            fmt.Printf("Erros: %d\n", count)
        } else {
            fmt.Printf("Status %d: %d\n", status, count)
        }
    }
    fmt.Printf("Tempo total: %v\n", time.Since(start))
}

Essa agregação nos dá uma visão geral da saúde dos endpoints. Em um cenário real, você poderia salvar esses dados em um arquivo JSON, enviar para um banco de dados ou gerar gráficos.

Erros

Lidar com erros é crucial em qualquer programa de rede. No nosso coletor, vários tipos de erro podem ocorrer: falha de conexão, timeout, resposta com status 5xx, etc. Vamos tratar cada um de forma adequada.

Primeiro, vamos melhorar a função fetch para usar um timeout e diferenciar erros de rede de erros de status. Também podemos classificar erros para uma melhor agregação.

type ErrorType int

const (
    ErrNetwork ErrorType = iota
    ErrTimeout
    ErrStatus
)

type FetchError struct {
    Type ErrorType
    Msg  string
}

func (e *FetchError) Error() string {
    return e.Msg
}

func fetchWithTimeout(url string, timeout time.Duration) Result {
    client := http.Client{Timeout: timeout}
    resp, err := client.Get(url)
    if err != nil {
        // Verifica se é timeout
        if os.IsTimeout(err) {
            return Result{URL: url, Err: &FetchError{Type: ErrTimeout, Msg: "timeout"}}
        }
        return Result{URL: url, Err: &FetchError{Type: ErrNetwork, Msg: err.Error()}}
    }
    defer resp.Body.Close()

    if resp.StatusCode < 200 || resp.StatusCode >= 300 {
        return Result{URL: url, StatusCode: resp.StatusCode, Err: &FetchError{Type: ErrStatus, Msg: fmt.Sprintf("status %d", resp.StatusCode)}}
    }

    // ... leitura do corpo
    return Result{URL: url, StatusCode: resp.StatusCode, Size: size}
}

No código acima, usamos os.IsTimeout para detectar timeouts. Essa função verifica se o erro é um timeout de rede. Para status não-2xx, retornamos um erro de status, mas ainda incluímos o código de status no Result para que possamos agregar.

No agregador, podemos contar erros por tipo:

func aggregate(results <-chan Result) (map[int]int, map[ErrorType]int) {
    statusCount := make(map[int]int)
    errorCount := make(map[ErrorType]int)
    for res := range results {
        if res.Err != nil {
            if fe, ok := res.Err.(*FetchError); ok {
                errorCount[fe.Type]++
            } else {
                errorCount[ErrNetwork]++
            }
            continue
        }
        statusCount[res.StatusCode]++
    }
    return statusCount, errorCount
}

Essa abordagem nos permite entender se os erros são de rede, timeout ou status, facilitando o diagnóstico.

Projeto completo: coletor concorrente

Agora vamos juntar tudo em um programa coeso. O projeto terá as seguintes funcionalidades:

  • Aceitar uma lista de URLs via argumentos de linha de comando.
  • Usar um número configurável de workers (padrão 3).
  • Limitar a taxa para 2 requisições por segundo.
  • Coletar resultados e gerar um relatório com contagens de status e erros.

Vamos escrever o código completo:

package main

import (
    "flag"
    "fmt"
    "net/http"
    "os"
    "sync"
    "time"
)

type ErrorType int

const (
    ErrNetwork ErrorType = iota
    ErrTimeout
    ErrStatus
)

type FetchError struct {
    Type ErrorType
    Msg  string
}

func (e *FetchError) Error() string {
    return e.Msg
}

type Result struct {
    URL        string
    StatusCode int
    Size       int
    Err        error
}

func fetchWithTimeout(url string, timeout time.Duration) Result {
    client := http.Client{Timeout: timeout}
    resp, err := client.Get(url)
    if err != nil {
        if os.IsTimeout(err) {
            return Result{URL: url, Err: &FetchError{Type: ErrTimeout, Msg: "timeout"}}
        }
        return Result{URL: url, Err: &FetchError{Type: ErrNetwork, Msg: err.Error()}}
    }
    defer resp.Body.Close()

    if resp.StatusCode < 200 || resp.StatusCode >= 300 {
        return Result{URL: url, StatusCode: resp.StatusCode, Err: &FetchError{Type: ErrStatus, Msg: fmt.Sprintf("status %d", resp.StatusCode)}}
    }

    buf := make([]byte, 1024)
    size := 0
    for {
        n, err := resp.Body.Read(buf)
        size += n
        if err != nil {
            break
        }
    }
    return Result{URL: url, StatusCode: resp.StatusCode, Size: size}
}

func worker(id int, urls <-chan string, results chan<- Result, limiter *time.Ticker, wg *sync.WaitGroup) {
    defer wg.Done()
    for url := range urls {
        <-limiter.C // limitação de taxa
        fmt.Printf("Worker %d: buscando %s\n", id, url)
        results <- fetchWithTimeout(url, 10*time.Second)
    }
}

func main() {
    numWorkers := flag.Int("workers", 3, "número de workers")
    rate := flag.Float64("rate", 2, "requisições por segundo")
    flag.Parse()

    urls := flag.Args()
    if len(urls) == 0 {
        fmt.Fprintln(os.Stderr, "Uso: coletor [opções] URL1 URL2 ...")
        os.Exit(1)
    }

    interval := time.Duration(1.0 / *rate * float64(time.Second))
    limiter := time.NewTicker(interval)
    defer limiter.Stop()

    urlsChan := make(chan string)
    resultsChan := make(chan Result)
    var wg sync.WaitGroup

    for i := 1; i <= *numWorkers; i++ {
        wg.Add(1)
        go worker(i, urlsChan, resultsChan, limiter, &wg)
    }

    go func() {
        defer close(urlsChan)
        for _, url := range urls {
            urlsChan <- url
        }
    }()

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

    statusCount := make(map[int]int)
    errorCount := make(map[ErrorType]int)
    totalBytes := 0
    start := time.Now()

    for res := range resultsChan {
        if res.Err != nil {
            if fe, ok := res.Err.(*FetchError); ok {
                errorCount[fe.Type]++
            } else {
                errorCount[ErrNetwork]++
            }
            fmt.Printf("Erro em %s: %v\n", res.URL, res.Err)
            continue
        }
        statusCount[res.StatusCode]++
        totalBytes += res.Size
        fmt.Printf("%s - Status: %d - Size: %d bytes\n", res.URL, res.StatusCode, res.Size)
    }

    fmt.Println("\nRelatório:")
    for status, count := range statusCount {
        fmt.Printf("Status %d: %d\n", status, count)
    }
    fmt.Printf("Total de bytes: %d\n", totalBytes)
    fmt.Printf("Erros de rede: %d\n", errorCount[ErrNetwork])
    fmt.Printf("Erros de timeout: %d\n", errorCount[ErrTimeout])
    fmt.Printf("Erros de status: %d\n", errorCount[ErrStatus])
    fmt.Printf("Tempo total: %v\n", time.Since(start))
}

Este código é um exemplo funcional de um coletor concorrente. Você pode compilar e executar com go run coletor.go -workers=5 -rate=3 https://httpbin.org/status/200 https://httpbin.org/status/404 etc.

Boas práticas e observações finais

Ao desenvolver sistemas concorrentes, sempre considere:

  • Contextos: Use context.Context para cancelamento e propagação de deadlines. Isso é essencial em aplicações reais.
  • Gerenciamento de recursos: Sempre feche corpos de resposta e canais.
  • Escalabilidade: Ajuste o número de workers e a taxa de acordo com os limites do serviço alvo.
  • Testes: Escreva testes para suas funções, especialmente para o tratamento de erros.
  • Observabilidade: Em produção, adicione logs estruturados e métricas.

Este projeto é uma base sólida para ferramentas de monitoramento, scraping e integração. Continue explorando e adaptando-o às suas necessidades.

Referências

Exercícios

  1. Modifique o número de workers para 5 e execute o programa com as URLs do exemplo. Observe como a saída muda em termos de ordem e tempo. Explique por que a ordem não é determinística.

    2. ✓ Resposta: Ao aumentar o número de workers, mais requisições são processadas simultaneamente, o que tende a reduzir o tempo total. A ordem de saída não é determinística porque as goroutines são agendadas de forma independente pelo runtime, e a conclusão de cada requisição depende de fatores como latência de rede e carga do sistema. Portanto, a ordem em que os resultados aparecem no canal pode variar a cada execução.
  2. Implemente um limite de taxa de 1 requisição por segundo e teste com 5 URLs. Meça o tempo total e compare com o caso sem limite (ou com limite maior).

    3. ✓ Resposta: Com 1 requisição por segundo e 5 URLs, o tempo total será de pelo menos 5 segundos (mais o tempo de cada requisição). Com 2 requisições por segundo, o tempo mínimo é de 2,5 segundos. A limitação de taxa é importante para não sobrecarregar o servidor, mas aumenta o tempo total. Você pode calcular o tempo mínimo como ceil(numero_de_urls / taxa).
  3. Adicione um campo Duration ao struct Result para armazenar o tempo de cada requisição. Modifique a função fetch para medir e preencher esse campo. No relatório, exiba a média de tempo por requisição.

    4. ✓ Resposta: Altere a struct Result para incluir Duration time.Duration. Na função fetchWithTimeout, registre o tempo antes e depois da chamada. No loop de coleta, acumule os tempos em uma variável e divida pelo número de resultados bem-sucedidos para obter a média. Exemplo: 
    start := time.Now()
resp, err := client.Get(url)
duration := time.Since(start)
// ...
return Result{URL: url, StatusCode: resp.StatusCode, Size: size, Duration: duration}
  4. Trate erros de forma mais granular: em vez de contar apenas o tipo de erro, conte também o código de status HTTP para erros de status. Modifique a agregação para mostrar, por exemplo, quantos 404 e quantos 500 ocorreram.

    5. ✓ Resposta: No caso de erro de status, o Result já contém o código de status. No agregador, se o erro for do tipo ErrStatus, incrementamos o contador correspondente ao código de status. Por exemplo: 
    if fe.Type == ErrStatus {
    statusCount[res.StatusCode]++
}
    Isso permite distinguir diferentes códigos de erro.
  5. Use context.Context para adicionar um timeout global ao programa (por exemplo, 30 segundos). Se o tempo for excedido, cancele todas as requisições em andamento e imprima uma mensagem de erro.

    6. ✓ Resposta: Crie um contexto com timeout: ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) e passe para as funções de requisição usando http.NewRequestWithContext. No worker, verifique se o contexto foi cancelado e pare de processar. No main, após o loop, verifique se o contexto expirou e imprima uma mensagem. Exemplo de requisição com contexto: 
    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
resp, err := client.Do(req)