您当前的位置：学无止境 > go简单操作kafka网站首页学无止境

go简单操作kafka

发布时间：2021-03-22 12:02:19编辑：sanqing查看次数：1034

启动kafka（zookeeper和kafka），参考https://lovesanqing.com/knowledge-181.html

下载sarama

go get github.com/Shopify/sarama

Windows平台请使用v1.19版本的sarama。（在go.mod修改版本号，再go mod tidy一下）

链接kafka消息


package main

import (
   "fmt"

   "github.com/Shopify/sarama"
)

// kafka consumer

func main() {
   consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
   if err != nil {
      fmt.Printf("fail to start consumer, err:%vn", err)
      return
   }
   partitionList, err := consumer.Partitions("sanqing") // 根据topic取到所有的分区
   if err != nil {
      fmt.Printf("fail to get list of partition:err%vn", err)
      return
   }
   fmt.Println("分区列表:",partitionList)
   for partition := range partitionList { // 遍历所有的分区
      // 针对每个分区创建一个对应的分区消费者
      pc, err := consumer.ConsumePartition("sanqing", int32(partition), sarama.OffsetNewest)
      if err != nil {
         fmt.Printf("failed to start consumer for partition %d,err:%vn", partition, err)
         return
      }
      defer pc.AsyncClose()
      // 异步从每个分区消费信息
      go func(sarama.PartitionConsumer) {
         for msg := range pc.Messages() {
            fmt.Printf("Partition:%d Offset:%d Key:%v Value:%vn", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
         }
      }(pc)
   }
   select {}
}

连接kafka发送消息

package main

import (
   "fmt"

   "github.com/Shopify/sarama"
)

// 基于sarama第三方库开发的kafka client

func main() {
   config := sarama.NewConfig()
   config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
   
   config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
   config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

   // 构造一个消息
   msg := &sarama.ProducerMessage{}
   msg.Topic = "sanqing"
   msg.Value = sarama.StringEncoder("哈哈 this is a test log")
   // 连接kafka
   client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
   if err != nil {
      fmt.Println("producer closed, err:", err)
      return
   }
   defer client.Close()
   // 发送消息
   pid, offset, err := client.SendMessage(msg)
   if err != nil {
      fmt.Println("send msg failed, err:", err)
      return
   }
   fmt.Printf("pid:%v offset:%vn", pid, offset)
}