Membina Pengeluar dan Pengguna Kafka dalam Go

Apache Kafka ialah platform penstriman teragih yang berkuasa yang digunakan untuk membina saluran paip data dan aplikasi penstriman masa nyata. Dalam catatan blog ini, kami akan meneruskan penyediaan pengeluar dan pengguna Kafka menggunakan Golang.

Prasyarat

Sebelum kami memulakan, pastikan anda telah memasang yang berikut pada mesin anda:

• Pergi (1.16 atau lebih tinggi)

• Docker (untuk menjalankan Kafka secara tempatan)

• Kafka

Menyediakan Kafka dengan Docker

Untuk menyediakan Kafka dengan cepat, kami akan menggunakan Docker. Buat fail docker-compose.yml dalam direktori projek anda:

yamlCopy codeversion: '3.7'
services:
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.13-2.7.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    depends_on:
      - zookeeper

Jalankan arahan berikut untuk memulakan Kafka dan Zookeeper:

docker-compose up -d

Mencipta Pengeluar Kafka dalam Go

Mula-mula, mulakan modul Go baharu:

go mod init kafka-example

Pasang perpustakaan kafka-go:

go get github.com/segmentio/kafka-go

Sekarang, buat fail producer.go dan tambah kod berikut:

package main

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

func main() {
    writer := kafka.Writer{
        Addr:     kafka.TCP("localhost:9092"),
        Topic:    "example-topic",
        Balancer: &kafka.LeastBytes{},
    }

    defer writer.Close()

    for i := 0; i < 10; i++ {
        msg := kafka.Message{
            Key:   []byte(fmt.Sprintf("Key-%d", i)),
            Value: []byte(fmt.Sprintf("Hello Kafka %d", i)),
        }

        err := writer.WriteMessages(context.Background(), msg)
        if err != nil {
            log.Fatal("could not write message " + err.Error())
        }

        time.Sleep(1 * time.Second)
        fmt.Printf("Produced message: %s\n", msg.Value)
    }
}

Kod ini menyediakan pengeluar Kafka yang menghantar sepuluh mesej ke topik topik contoh.

Jalankan penerbit:

go run producer.go

Anda sepatutnya melihat output yang menunjukkan bahawa mesej telah dihasilkan.

Mencipta Pengguna Kafka dalam Go

Buat fail consumer.go dan tambah kod berikut:

package main

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

func main() {
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"localhost:9092"},
        Topic:   "example-topic",
        GroupID: "example-group",
    })

    defer reader.Close()

    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal("could not read message " + err.Error())
        }
        fmt.Printf("Consumed message: %s\n", msg.Value)
    }
}

Pengguna ini membaca mesej daripada topik topik contoh dan mencetaknya ke konsol.

Jalankan pengguna:

go run consumer.go

Anda sepatutnya melihat output yang menunjukkan bahawa mesej telah digunakan.

Kesimpulan

Dalam catatan blog ini, kami menunjukkan cara menyediakan pengeluar dan pengguna Kafka menggunakan Golang. Contoh mudah ini menunjukkan asas menghasilkan dan menggunakan mesej, tetapi keupayaan Kafka melampaui ini. Dengan Kafka, anda boleh membina sistem pemprosesan data masa nyata yang teguh dan berskala.

Jangan ragu untuk meneroka ciri yang lebih maju seperti pembahagian mesej, pengedaran mesej berasaskan kunci dan penyepaduan dengan sistem lain. Selamat mengekod!

---

Itu sahaja! Catatan blog ini menyediakan pengenalan ringkas untuk menggunakan Kafka dengan Go, sesuai untuk pembangun yang ingin bermula dengan pemprosesan data masa nyata.

Atas ialah kandungan terperinci Membina Pengeluar dan Pengguna Kafka dalam Go. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!