Trabalhando com Apache Kafka em Go (Golang): bibliotecas e exemplos

Trabalhando com Apache Kafka em Go (Golang): bibliotecas, exemplos e melhores práticas

Apache Kafka é uma plataforma distribuída para streaming de dados que se tornou o padrão de facto para a construção de sistemas de troca de mensagens de alta carga. No ecossistema Go (Golang), existem várias bibliotecas maduras para interagir com Kafka. Neste artigo, vamos analisar as duas mais populares: confluent-kafka-go (cliente oficial da Confluent) e sarama (cliente Go puro). Você aprenderá como configurar um produtor e um consumidor, tratar erros e escolher a ferramenta adequada para o seu projeto.

Por que Kafka e Go são uma excelente combinação?

Go foi criado para desenvolver aplicações server-side rápidas, concorrentes e confiáveis. Kafka, por sua vez, exige alta taxa de transferência e latência mínima. As goroutines e canais do Go são ideais para o processamento assíncrono de fluxos de mensagens. Além disso, a tipagem estática do Go ajuda a evitar muitos erros em tempo de compilação, o que é crítico para sistemas em produção.

1. Biblioteca confluent-kafka-go

Este é um wrapper sobre o cliente C librdkafka, oferecendo desempenho máximo e suporte completo a todos os recursos do Kafka (transações, idempotência, protocolos de segurança). É recomendado para sistemas de alta carga.

Instalação

Requer a presença do librdkafka no sistema:

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# Em seguida, instalar o pacote Gogo get -u github.com/confluentinc/confluent-kafka-go/kafka

Exemplo de Produtor (Producer)

Envia uma mensagem para o tópico orders:

package main

import (    "fmt"    "github.com/confluentinc/confluent-kafka-go/kafka")

func main() {    p, err := kafka.NewProducer(&kafka.ConfigMap{        "bootstrap.servers": "localhost:9092",        "acks":              "all",    })    if err != nil {        panic(err)    }    defer p.Close()

    // Canal para notificações de entrega    go func() {        for e := range p.Events() {            switch ev := e.(type) {            case *kafka.Message:                if ev.TopicPartition.Error != nil {                    fmt.Printf("Erro de entrega: %v", ev.TopicPartition.Error)                } else {                    fmt.Printf("Mensagem entregue em %s [%d] offset %d",                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)                }            }        }    }()

    topic := "orders"    message := []byte("Novo pedido #12345")    err = p.Produce(&kafka.Message{        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},        Value:          message,        Key:            []byte("order_key"),    }, nil)

    if err != nil {        fmt.Printf("Erro ao enviar: %v", err)    }

    // Aguarda o envio de todas as mensagens    p.Flush(15 * 1000)}

Exemplo de Consumidor (Consumer)

Lê mensagens do tópico orders no grupo order-processors:

package main

import (    "fmt"    "github.com/confluentinc/confluent-kafka-go/kafka")

func main() {    c, err := kafka.NewConsumer(&kafka.ConfigMap{        "bootstrap.servers": "localhost:9092",        "group.id":          "order-processors",        "auto.offset.reset": "earliest",    })    if err != nil {        panic(err)    }    defer c.Close()

    err = c.SubscribeTopics([]string{"orders"}, nil)    if err != nil {        panic(err)    }

    for {        msg, err := c.ReadMessage(-1) // espera infinita        if err != nil {            fmt.Printf("Erro ao receber: %v", err)            continue        }        fmt.Printf("Mensagem recebida: %s (chave: %s) de %s [%d] of