GO 语言操作 kafka

GO 语言操作 kafka

kafka 提供了多种客户端，可以使用 segmentio/kafka-go 包快速操作 kafka。

segmentio/kafka-go 官网

https://pkg.go.dev/github.com/segmentio/kafka-go

示例代码

说明 : 需要安装并启动 kafka 环境。

package main

import (
	"context"
	"fmt"
	"log"
	"strings"

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

func getKafkaReader(kafkaURL, topic, groupID string) *kafka.Reader {
	brokers := strings.Split(kafkaURL, ",")
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers:  brokers,
		GroupID:  groupID,
		Topic:    topic,
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
	})
}

func main() {
	// get kafka reader using environment variables.
	//kafkaURL := os.Getenv("kafkaURL")
	kafkaURL := "192.168.1.101:9092"
	//topic := os.Getenv("topic")
	topic := "first"
	//groupID := os.Getenv("groupID")
	groupID := ""

	reader := getKafkaReader(kafkaURL, topic, groupID)

	defer reader.Close()

	fmt.Println("start consuming ... !!")
	for {
		m, err := reader.ReadMessage(context.Background())
		if err != nil {
			log.Fatalln(err)
		}
		fmt.Printf("message at topic:%v partition:%v offset:%v	%s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
	}
}