Go 操作kafka_go 消费kafka

1.下载和安装

go get  github.com/IBM/sarama

2.连接

func main() {
   config := sarama.NewConfig()
   config.Producer.RequiredAcks = sarama.WaitForAll //发送完数据需要leader和follow都确认
   config.Producer.Partitioner = sarama.NewRandomPartitioner
   config.Producer.Return.Successes = true         // 成功交付的消息将在success channel返回
   config.Producer.Retry.Max = 0                   //重新发送的次数
   config.Producer.Timeout = time.Microsecond * 10 //等待 WaitForAck的时间
   config.Producer.Idempotent = true               //开启幂等性
   //config.Producer.Transaction =

   //构造一个消息

   //连接kafka
   client, err := sarama.NewSyncProducer([]string{"192.168.0.4:9092"}, config)
   if err != nil {
      fmt.Println("producer closed,err:", err)
      return
   }
   defer client.Close()

   //发送消息
   for i := 0; i < 5; i++ {
      msg := &sarama.ProducerMessage{}
      msg.Topic = "user-top"
      msg.Value = sarama.StringEncoder("this is test log:" + strconv.Itoa(i))
      pid, offset, err := client.SendMessage(msg)
      if err != nil {
         fmt.Println("send msg failed,err:", err)
         return
      }
      fmt.Printf("pid:%v offset:%v\n", pid, offset)
   }

}

3.连接kafka消费消息

func main() {
   consumer, err := sarama.NewConsumer([]string{"192.168.0.4:9092"}, nil)
   if err != nil {
      fmt.Println("err1", err)
      fmt.Printf("fail to start consumer,err:%v\n", err)
      return
   }
   partitionList, err := consumer.Partitions("user-topic")
   if err != nil {
      fmt.Println("err2", err)
      fmt.Printf("fail to get list of partition:err%v\n", err)
      return
   }
   for _, partition := range partitionList {
      // 针对每个分区创建一个对应的分区消费者
      pc, err := consumer.ConsumePartition("user-topic", int32(partition), sarama.OffsetNewest)
      if err != nil {
         fmt.Println("err3", err)
         fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
         return
      }
      defer pc.AsyncClose()
      // 异步从每个分区消费信息
      go func(sarama.PartitionConsumer) {
         for msg := range pc.Messages() {
            fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
         }
      }(pc)
   }
}