🛠 Как использовать кафку в Go

Представьте, что у вас в руках инструмент, который позволяет строить крутой обмен сообщениями между сервисами. Представили? Давайте попробуем его применить.

1. Установка

Для начала проверьте, что у вас установлен Go версии 1.15 и выше и введите команду:

go get github.com/segmentio/kafka-go

import "github.com/segmentio/kafka-go"

ctx := context.Background()
conn, err := kafka.DialLeader(ctx, "tcp", "localhost:9092", "my-topic", 0)
if err != nil {
    log.Fatal(err)
}
defer conn.Close()

// Отправляем два сообщения подряд
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err = conn.WriteMessages(
    kafka.Message{Value: []byte("Первое сообщение")},
    kafka.Message{Value: []byte("Второе сообщение")},
)
if err != nil {
    log.Fatal(err)
}

batch := conn.ReadBatch(10e3, 1e6)
defer batch.Close()

buf := make([]byte, 10e3)
for {
    n, err := batch.Read(buf)
    if err != nil {
        break
    }
    fmt.Println("→", string(buf[:n]))
}

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092", "localhost:9093"},
    GroupID:        "example-group",
    Topic:          "events",
    MinBytes:       10e3,
    MaxBytes:       10e6,
    CommitInterval: time.Second,
})
defer r.Close()

for {
    m, err := r.ReadMessage(context.Background())
    if err != nil {
        log.Println("Завершение чтения:", err)
        break
    }
    fmt.Printf("Получено: ключ=%s, значение=%s, смещение=%d\n", m.Key, m.Value, m.Offset)
}

w := kafka.NewWriter(kafka.WriterConfig{
    Brokers:   []string{"localhost:9092", "localhost:9093"},
    Topic:     "events",
    Balancer:  &kafka.LeastBytes{},
})
defer w.Close()

messages := []kafka.Message{
    {Key: []byte("order123"), Value: []byte(`{"status":"created"}`)},
    {Key: []byte("order124"), Value: []byte(`{"status":"paid"}`)},
}
if err := w.WriteMessages(context.Background(), messages...); err != nil {
    log.Fatal("Ошибка при записи:", err)
}