golang之操作kafka

安装第三方包:

go get github.com/IBM/sarama

生产者实例:

package main

import (
   "fmt"
   "github.com/IBM/sarama"
)

func main() {

   //1.生产者配置
   config := sarama.NewConfig()
   config.Producer.RequiredAcks = sarama.WaitForAll          //ACK,发送完数据需要leader和follow都确认
   config.Producer.Partitioner = sarama.NewRandomPartitioner //分区,新选出一个分区
   config.Producer.Return.Successes = true                   //确认,成功交付的消息将在success channel返回

   //2.连接Kafka

   client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
   if err != nil {
      fmt.Println("Producer error", err)
      return
   }

   defer client.Close()

   //3.封装消息

   msg := &sarama.ProducerMessage{}
   msg.Topic = "log"
   msg.Value = sarama.StringEncoder("this is test log")

   //4.发送消息
   pid, offset, err := client.SendMessage(msg)

   if err != nil {
      fmt.Println("send faild", err)
   }
   fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

消费者: