一文详解Golang连接kafka的基本操作

bin/zookeeper-server-start.sh config/zookeeper.properties

bin/kafka-server-start.sh config/server.properties

bin/kafka-topics.sh --create --topic hello --bootstrap-server 主机名:9092

bin/kafka-console-producer.sh --broker-list 主机名:9092 --topic hello

bin/kafka-console-consumer.sh --bootstrap-server 主机名:9092 --from-beginning --topic hello

bin/kafka-console-consumer.sh --bootstrap-server 主机名:9092 --partition 0 --offset 1 --topic hello

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic hello

// Kafka 配置
const (
	KafkaBroker = "u8sMaster:9092" // 替换为你的 Kafka Broker 地址
	KafkaTopic  = "k8s-version"          // Kafka 主题
)

func main() {
	sendMesgKafka()
}

func sendMesgKafka() {
	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers:  []string{KafkaBroker},
		Topic:    KafkaTopic,
		Balancer: &kafka.LeastBytes{},
	})

	err := w.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte("Key-A"),
			Value: []byte("one!"),
		},
		kafka.Message{
			Key:   []byte("Key-B"),
			Value: []byte("two!"),
		},
		kafka.Message{
			Key:   []byte("Key-C"),
			Value: []byte("three!"),
		},
	)

	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	if err := w.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}

	fmt.Println("Message sent successfully")

}

// to consume messages
topic := "test"
partition := 0

conn, err := kafka.DialLeader(context.Background(), "tcp", "u8sMaster:9092", topic, partition)
if err != nil {
    log.Fatal("failed to dial leader:", err)
}

conn.SetReadDeadline(time.Now().Add(10*time.Second))
batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

b := make([]byte, 10e3) // 10KB max per message
for {
    n, err := batch.Read(b)
    if err != nil {
        break
    }
    fmt.Println(string(b[:n]))
}

if err := batch.Close(); err != nil {
    log.Fatal("failed to close batch:", err)
}

if err := conn.Close(); err != nil {
    log.Fatal("failed to close connection:", err)
}

func main() {
    conn, err := kafka.Dial("tcp", "u8sMaster:9092")
    if err != nil {
        panic(err.Error())
    }
    defer conn.Close()
    
    partitions, err := conn.ReadPartitions()
    if err != nil {
        panic(err.Error())
    }
    
    m := map[string]struct{}{}
    
    for _, p := range partitions {
        m[p.Topic] = struct{}{}
    }
    for k := range m {
        fmt.Println(k)
    }
}

func main() {
        conn, err := kafka.DialLeader(context.Background(), "tcp", "u9sMaster:9092", "topic2", 0)
        if err != nil {
            panic(err.Error())
        }
}

func main() {
    conn, err := kafka.Dial("tcp", "u8sMaster:9092")
    if err != nil {
        panic(err.Error())
    }
    defer conn.Close()
    controller, err := conn.Controller()
    if err != nil {
        panic(err.Error())
    }
    var connLeader *kafka.Conn
    connLeader, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
    if err != nil {
        panic(err.Error())
    }
    defer connLeader.Close()
}