1.kafka的学习


1.1 启动kafka与zookeeper

kafka与zookeeper是相关联的

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

复制代码

与

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

复制代码

1.2 创建topic

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

复制代码

1.3 生产消息

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

复制代码

运行后可以发送多条,ctrl+c退出

1.4 消费之前的消息

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

复制代码

1.5 指定偏移量消费

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

复制代码

1.6 消费最新的信息

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

复制代码

2 go操作


2.1 发送消息

1. // Kafka 配置
   
2. const (
   
3.         KafkaBroker = "u8sMaster:9092" // 替换为你的 Kafka Broker 地址
   
4.         KafkaTopic  = "k8s-version"          // Kafka 主题
   
5. )
   
6. 
   
7. func main() {
   
8.         sendMesgKafka()
   
9. }
   
10. 
    
11. func sendMesgKafka() {
    
12.         w := kafka.NewWriter(kafka.WriterConfig{
    
13.                 Brokers:  []string{KafkaBroker},
    
14.                 Topic:    KafkaTopic,
    
15.                 Balancer: &kafka.LeastBytes{},
    
16.         })
    
17. 
    
18.         err := w.WriteMessages(context.Background(),
    
19.                 kafka.Message{
    
20.                         Key:   []byte("Key-A"),
    
21.                         Value: []byte("one!"),
    
22.                 },
    
23.                 kafka.Message{
    
24.                         Key:   []byte("Key-B"),
    
25.                         Value: []byte("two!"),
    
26.                 },
    
27.                 kafka.Message{
    
28.                         Key:   []byte("Key-C"),
    
29.                         Value: []byte("three!"),
    
30.                 },
    
31.         )
    
32. 
    
33.         if err != nil {
    
34.                 log.Fatal("failed to write messages:", err)
    
35.         }
    
36. 
    
37.         if err := w.Close(); err != nil {
    
38.                 log.Fatal("failed to close writer:", err)
    
39.         }
    
40. 
    
41.         fmt.Println("Message sent successfully")
    
42. 
    
43. }

复制代码

2.2 消费消息

1. // to consume messages
   
2. topic := "test"
   
3. partition := 0
   
4. 
   
5. conn, err := kafka.DialLeader(context.Background(), "tcp", "u8sMaster:9092", topic, partition)
   
6. if err != nil {
   
7.     log.Fatal("failed to dial leader:", err)
   
8. }
   
9. 
   
10. conn.SetReadDeadline(time.Now().Add(10*time.Second))
    
11. batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max
    
12. 
    
13. b := make([]byte, 10e3) // 10KB max per message
    
14. for {
    
15.     n, err := batch.Read(b)
    
16.     if err != nil {
    
17.         break
    
18.     }
    
19.     fmt.Println(string(b[:n]))
    
20. }
    
21. 
    
22. if err := batch.Close(); err != nil {
    
23.     log.Fatal("failed to close batch:", err)
    
24. }
    
25. 
    
26. if err := conn.Close(); err != nil {
    
27.     log.Fatal("failed to close connection:", err)
    
28. }

复制代码

2.3 列出所有topic

1. func main() {
   
2.     conn, err := kafka.Dial("tcp", "u8sMaster:9092")
   
3.     if err != nil {
   
4.         panic(err.Error())
   
5.     }
   
6.     defer conn.Close()
   
7.    
   
8.     partitions, err := conn.ReadPartitions()
   
9.     if err != nil {
   
10.         panic(err.Error())
    
11.     }
    
12.    
    
13.     m := map[string]struct{}{}
    
14.    
    
15.     for _, p := range partitions {
    
16.         m[p.Topic] = struct{}{}
    
17.     }
    
18.     for k := range m {
    
19.         fmt.Println(k)
    
20.     }
    
21. }

复制代码

2.4 创建topic

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

复制代码

精准地创建topic

1. func main() {
   
2.     conn, err := kafka.Dial("tcp", "u8sMaster:9092")
   
3.     if err != nil {
   
4.         panic(err.Error())
   
5.     }
   
6.     defer conn.Close()
   
7.     controller, err := conn.Controller()
   
8.     if err != nil {
   
9.         panic(err.Error())
   
10.     }
    
11.     var connLeader *kafka.Conn
    
12.     connLeader, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
    
13.     if err != nil {
    
14.         panic(err.Error())
    
15.     }
    
16.     defer connLeader.Close()
    
17. }

复制代码

这里省略了kafka集群的配置，未来有机会补充
以上就是一文详解Golang连接kafka的基本操作的详细内容，更多关于go连接kafka的资料请关注脚本之家其它相关文章！

来源：互联网
免责声明：如果侵犯了您的权益，请联系站长（1277306191@qq.com），我们会及时删除侵权内容，谢谢合作！