go与kafka(github.com/Shopify/sarama)

本文主要是介绍go与kafka(github.com/Shopify/sarama)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

启动kafka

分别命令行启动zookeeper和kafka

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
.\bin\windows\kafka-server-start.bat .\config\server.properties

 producer

import ("fmt""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()config.Producer.RequiredAcks = sarama.WaitForAll  //赋值为-1：这意味着producer在follower副本确认接收到数据后才算一次发送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner  //写到随机分区中，默认设置8个分区config.Producer.Return.Successes = truemsg := &sarama.ProducerMessage{}msg.Topic = `nginx_log`msg.Value = sarama.StringEncoder("this is a good test")client,err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"},config)if err != nil{fmt.Println("producer close err, ",err)return}defer client.Close()pid, offset, err := client.SendMessage(msg)if err != nil{fmt.Println("send message failed, ", err)return}fmt.Printf("分区ID:%v, offset:%v \n", pid, offset)}

consumer

func main() {var wg sync.WaitGroupconsumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)if err != nil{fmt.Println("Failed to start consumer: %s", err)return}partitionList, err := consumer.Partitions("nginx_log")  //获得该topic所有的分区if err != nil{fmt.Println("Failed to get the list of partition:, ",err)return}fmt.Println(partitionList)for partition := range partitionList{pc, err := consumer.ConsumePartition("nginx_log", int32(partition), sarama.OffsetNewest)if err != nil{fmt.Println("Failed to start consumer for partition %d: %s\n", partition, err)return}wg.Add(1)go func(sarama.PartitionConsumer) { //为每个分区开一个go协程去取值for msg := range pc.Messages(){  //阻塞直到有值发送过来，然后再继续等待fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset, string(msg.Key),string(msg.Value))}defer pc.AsyncClose()wg.Done()}(pc)}wg.Wait()consumer.Close()
}

 

这篇关于go与kafka(github.com/Shopify/sarama)的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---