Golang操作kafka

2021-11-11 22:03:09 阅读：361 来源： 互联网

标签：err sarama 分区 partition kafka Golang 操作 leader

kafka cluster: kafka集群，一台或多台服务器组成
        Broker: Broker是指部署了Kafka实例的服务器节点。每个服务器上有一个或多个kafka的实
例，我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1等......
        Topic:消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。实际应用中通常是一个业务线建一个topic。
        Partition: Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的,partition的表现形式就是一个一个的文件夹!
        Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区(Leader)故障的
时候会选择一个备胎（Follower)上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本(包括自己)。

consumer:消费者，即消息的消费方，是消息的出口。
Consumer Group:我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分
区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量!
工作流程
我们看上面的架构图中，producer就是生产者，是数据的入口。Producer在写入数据的时候会把数据写入到leader中，不会直接将数据写入follower!那leader怎么找呢?写入的流程又是什么样的呢?我们看下图:

1.Kafka集群的架构
        1.broker: 节点
        2.topic: 消息主题
        3.partition: 分区，把同一个topic分成不同的分区，提高负载
                1.leader: 分区的主节点(老大)
                2. follower: 分区的从节点（小弟)
        4.Consumer Group
2.生产者往Kafka发送数据的流程（6步)
        1. 生产者从Kafka集群获取分区leader信息
        2. 生产者将消息发送给leader
        3. leader将消息写入本地磁盘
        4. follower从leader拉取消息数据
        5. follower将消息写入本地磁盘后向leader发送ACK
        6. leader收到所有的follower的ACK之后向生产者发送ACK
3.Kafka选择分区的模式(3种)
        1.指定往哪个分区写
        2.指定key，kafka根据key做hash然后决定写哪个分区
        3.轮询方式
4.生产者往kafka发送数据的模式(3种)
        1. 0 :把数据发给leader就成功，效率最高、安全性最低。
        2. 1 :把数据发送给leader，等待leader回ACK
        3. all :把数据发给leader,确保follower从leader拉取数据回复ack给leader，leader再回复ACK，安全性最高

选择partition的原则

        那在kafka中，如果某个topic有多个partition，producer又怎么知道该将数据发往哪个partition呢?kafka中有几个原则:
        1. partition在写入的时候可以指定需要写入的partition，如果有指定，则写入对应的partition。
2.如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。
        3.如果既没指定partition，又没有设置key，则会采用轮询方式，即每次取一小段时间的数据写入某个partition，下一小段的时间写入下一个partition。

下载并启动kafka，需先配置jdk环境

下载好kafka后修改zookeeper.properties和server.properties两个配置文件的日志生成路径

启动

linux和mac执行 bin下的kafka-server-start.sh和zookeeper-server-start.sh

windows以管理员身份打开cmd终端执行bin下的windows下的kafka-server-start.bat和zookeeper-server-start.bat

运行并指定上面修改的配置文件，先启动zookeeper

LogAgent的工作流程:

1.读日志 --tailf第三方库
2.往kafka写日志

1.读日志，使用tailf库

go get github.com/hpcloud/tail

tailf文档代码示范

package main

import (
   "fmt"
   "github.com/hpcloud/tail"
   "time"
)

func main() {
   filename := "./my.log"
   config := tail.Config{
      ReOpen:    true,  //重新打开
      Follow:    true, // 是否跟随
      Location:    &tail.SeekInfo{Offset: 0, Whence: 2}, //从文件哪个地方开始读
      MustExist: false,              //文件不存在不报错
      Poll:      true,          //
   }
   tails,err := tail.TailFile(filename,config)
   if err != nil {
      fmt.Println("tail file failed, err",err)
      return
   }
   var (
      line *tail.Line
      ok bool
   )
   for {
      line,ok = <- tails.Lines
      if !ok {
         fmt.Printf("tail file close reopen, filename:%s\n",tails.Filename)
         time.Sleep(time.Second)
         continue
      }
      fmt.Println("line:",line.Text)
   }
}

log agent开发

sarama v1.20之后的版本加入了zstd压缩算法，需要用到cgo，在Windows平台编译时会提示类似如下错误:

github.com/ DataDog/zstd
exec: "gcc" :executable file not found in %PATH%

所以在Windows平台请使用v1.19版本的sarama。
在go.mod中写入

module kafka/kafka_demo //创建的包名 根据自己需求更改

go 1.17

require (                          
   github.com/Shopify/sarama v1.19  //windows指定版本  mac，linux不用  
                                    
)

然后打开在终端执行go mod download下载，linux，mac可直接go get github.com/Shopify/sarama下载。

go get github.com/Shopify/sarama

文档代码示范

package main

import (
   "fmt"
   "github.com/Shopify/sarama"
)
//基于sarama第三方库开发的kafka client
func main() {
   config := sarama.NewConfig()
   //tailf包使用
   config.Producer.RequiredAcks = sarama.WaitForAll //Producer生产者    发送完数据需要leader和follow都确认
   config.Producer.Partitioner = sarama.NewRandomPartitioner //Partitioner分区   新选出一个partition
   config.Producer.Return.Successes = true //成功交付的消息将在success channel返回
   //构造消息
   msg := &sarama.ProducerMessage{}
   msg.Topic = "web_log"
   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
   }
   fmt.Println("链接kafka成功！")
   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:%v\n",pid, offset)

}

执行会在kafka日志配置文件生成web_log文件夹

标签：err,sarama,分区,partition,kafka,Golang,操作,leader
来源： https://blog.csdn.net/qq_36410466/article/details/121276463

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Golang操作kafka

选择partition的原则

下载并启动kafka，需先配置jdk环境

LogAgent的工作流程: