Sarama para Go: guia completo com exemplos

O que é Sarama e para que serve?

\n\n

Sarama é uma biblioteca Go pura e de alto desempenho para trabalhar com Apache Kafka. Ela fornece um conjunto completo de ferramentas para interagir com um cluster Kafka, incluindo produtores (producers) para enviar mensagens, consumidores (consumers) para ler mensagens, além de funções administrativas para gerenciar tópicos, partições e configurações.

\n\n

Apache Kafka é uma plataforma distribuída de streaming de dados usada para construir pipelines em tempo real, sistemas de troca de mensagens, processamento de eventos e armazenamento de logs. Sarama permite que aplicações Go se integrem facilmente com Kafka, garantindo uma transmissão de dados confiável e eficiente.

\n\n

Por que usar Sarama?

\n

\n
• Go puro: Sarama é escrita inteiramente em Go, sem dependências externas de C/C++, simplificando a compilação e implantação.
\n
• Alto desempenho: A biblioteca é otimizada para trabalhar com grandes volumes de dados e suporta envio assíncrono de mensagens.
\n
• Flexibilidade: Suporta diferentes modos de operação para produtores (síncrono, assíncrono) e consumidores (simples, grupo de consumidores).
\n
• Comunidade ativa: Sarama é uma das bibliotecas Kafka mais populares para Go, com muitos contribuidores e documentação extensa.
\n
• Controle total: Dá ao desenvolvedor controle total sobre a configuração, incluindo gerenciamento de offsets, serialização/desserialização e tratamento de erros.
\n

\n\n

Instalação

\n\n

A instalação do Sarama em um projeto Go é feita da forma padrão através do gerenciador de pacotes go get.

\n\n

go get github.com/IBM/sarama

\n\n

Após executar este comando, a biblioteca será adicionada ao seu arquivo go.mod, e você poderá importá-la em seu código.

\n\n

Importante: Certifique-se de ter o Go versão 1.18 ou superior instalado, pois o Sarama utiliza recursos modernos da linguagem.

\n\n

Início rápido: enviando e recebendo mensagens

\n\n

Vamos escrever um exemplo simples que enviará uma mensagem para um tópico Kafka e a lerá.

\n\n

1. Produtor (Producer) — enviando mensagem

\n\n

package main\n\nimport (\n    "fmt"\n    "log"\n\n    "github.com/IBM/sarama"\n)\n\nfunc main() {\n    // 1. Configuração do produtor\n    config := sarama.NewConfig()\n    config.Producer.Return.Successes = true // Necessário para modo síncrono\n    config.Producer.RequiredAcks = sarama.WaitForAll // Aguardar confirmação de todas as réplicas\n\n    // 2. Criando produtor síncrono\n    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)\n    if err != nil {\n        log.Fatalf("Falha ao criar produtor: %v", err)\n    }\n    defer producer.Close()\n\n    // 3. Criando mensagem\n    msg := &sarama.ProducerMessage{\n        Topic: "my-topic",\n        Key:   sarama.StringEncoder("my-key"),\n        Value: sarama.StringEncoder("Olá, Kafka!"),\n    }\n\n    // 4. Enviando mensagem\n    partition, offset, err := producer.SendMessage(msg)\n    if err != nil {\n        log.Fatalf("Falha ao enviar mensagem: %v", err)\n    }\n\n    fmt.Printf("Mensagem enviada para partição %d com offset %d\\n", partition, offset)\n}\n

\n\n

2. Consumidor (Consumer) — recebendo mensagem

\n\n

package main\n\nimport (\n    "fmt"\n    "log"\n    "os"\n    "os/signal"\n    "syscall"\n\n    "github.com/IBM/sarama"\n)\n\nfunc main() {\n    // 1. Configuração do consumidor\n    config := sarama.NewConfig()\n    config.Consumer.Return.Errors = true\n    config.Consumer.Offsets.Initial = sarama.OffsetOldest // Ler desde o início\n\n    // 2. Criando consumidor\n    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)\n    if err != nil {\n        log.Fatalf("Falha ao criar consumidor: %v", err)\n    }\n    defer consumer.Close()\n\n    // 3. Consumindo mensagens da partição\n    partitionConsumer, err := consumer.ConsumePartition("my-topic", 0, sarama.OffsetOldest)\n    if err != nil {\n        log.Fatalf("Falha ao iniciar consumidor de partição: %v", err)\n    }\n    defer partitionConsumer.Close()\n\n    // 4. Capturando sinal para encerramento gracioso\n    signals := make(chan os.Signal, 1)\n    signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)\n\n    // 5. Loop de consumo\n    for {\n        select {\n        case msg := <-partitionConsumer.Messages():\n            fmt.Printf("Mensagem recebida: chave=%s, valor=%s, partição=%d, offset=%d\\n",\n                string(msg.Key), string(msg.Value), msg.Partition, msg.Offset)\n        case err := <-partitionConsumer.Errors():\n            log.Printf("Erro no consumidor: %v\\n", err)\n        case <-signals:\n            fmt.Println("Encerrando consumidor...")\n            return\n        }\n    }\n}\n