Trabajo con Apache Kafka en Go (Golang): bibliotecas y ejemplos

Trabajo con Apache Kafka en Go (Golang): bibliotecas, ejemplos y mejores prácticas

Apache Kafka es una plataforma distribuida para transmisión de datos que se ha convertido en el estándar de facto para construir sistemas de mensajería de alta carga. En el ecosistema de Go (Golang) existen varias bibliotecas maduras para interactuar con Kafka. En este artículo analizaremos las dos más populares: confluent-kafka-go (cliente oficial de Confluent) y sarama (cliente puro en Go). Aprenderá cómo configurar un productor y un consumidor, manejar errores y elegir la herramienta adecuada para su proyecto.

¿Por qué Kafka y Go son una excelente combinación?

Go fue creado para desarrollar aplicaciones de servidor rápidas, concurrentes y confiables. Kafka, por su parte, requiere alto rendimiento y mínima latencia. Las gorutinas y canales de Go son ideales para el procesamiento asíncrono de flujos de mensajes. Además, el tipado estático de Go ayuda a evitar muchos errores en tiempo de compilación, lo cual es crítico para sistemas en producción.

1. Biblioteca confluent-kafka-go

Es un envoltorio sobre el cliente C librdkafka, que ofrece el máximo rendimiento y soporte completo de todas las funcionalidades de Kafka (transacciones, idempotencia, protocolos de seguridad). Se recomienda para sistemas de alta carga.

Instalación

Requiere tener librdkafka instalado en el sistema:

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# Luego instalar el paquete Gogo get -u github.com/confluentinc/confluent-kafka-go/kafka

Ejemplo de productor (Producer)

Envía un mensaje al 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 notificaciones de entrega    go func() {        for e := range p.Events() {            switch ev := e.(type) {            case *kafka.Message:                if ev.TopicPartition.Error != nil {                    fmt.Printf("Error de entrega: %v", ev.TopicPartition.Error)                } else {                    fmt.Printf("Mensaje entregado en %s [%d] offset %d",                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)                }            }        }    }()

    topic := "orders"    message := []byte("Nuevo 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("Error al enviar: %v", err)    }

    // Esperar a que todos los mensajes se envíen    p.Flush(15 * 1000)}

Ejemplo de consumidor (Consumer)

Lee mensajes del tópico orders en el 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("Error al recibir: %v", err)            continue        }        fmt.Printf("Mensaje recibido: %s (clave: %s) de %s [%d] offset %d",            string(msg.Value), string(msg.Key), *msg.TopicPartition.Topic, msg.TopicPartition.Partition, msg.TopicPartition.Offset)    }}