Kafka with Go

Go Kafka Libraries

go get github.com/segmentio/kafka-go

Producer Example

package main

import (
    "context"
    "fmt"
    "github.com/segmentio/kafka-go"
)

func main() {
    writer := &kafka.Writer{
        Addr:         kafka.TCP("localhost:9092"),
        Topic:        "user-events",
        Balancer:     &kafka.LeastBytes{},
        RequiredAcks: kafka.RequireAll,
        Compression:  kafka.Lz4,
    }
    defer writer.Close()

    err := writer.WriteMessages(context.Background(),
        kafka.Message{Key: []byte("user_1"), Value: []byte(`{"action":"login"}`)},
        kafka.Message{Key: []byte("user_2"), Value: []byte(`{"action":"signup"}`)},
    )
    if err != nil {
        fmt.Printf("Failed to write: %v\n", err)
    }
    fmt.Println("Messages sent!")
}

Consumer Group Example

package main

import (
    "context"
    "fmt"
    "github.com/segmentio/kafka-go"
)

func main() {
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers:  []string{"localhost:9092"},
        GroupID:  "go-consumer",
        Topic:    "user-events",
        MinBytes: 1,
        MaxBytes: 10e6,
    })
    defer reader.Close()

    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            fmt.Printf("Error: %v\n", err)
            break
        }
        fmt.Printf("Partition: %d, Offset: %d, Key: %s, Value: %s\n",
            msg.Partition, msg.Offset, msg.Key, msg.Value)
    }
}

Practice Exercises