版权声明:本文为博主原创文章,转载注明出处,谢谢支持 https://blog.csdn.net/qq_32292967/article/details/78675116
概述
go语言版本的kafakaAPI
准备
简单使用
生产和消费的过程都是通过一个配置开始的.
生产者
//设置配置
config := sarama.NewConfig ()
//等待服务器所有副本都保存成功后的响应
config.Producer .RequiredAcks = sarama.WaitForAll
//随机的分区类型
config.Producer .Partitioner = sarama.NewRandomPartitioner
//是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
config.Producer .Return .Successes = true
config.Producer .Return .Errors = true
//设置使用的kafka版本,如果低于V0_10_0_0版本,消息中的timestrap没有作用.需要消费和生产同时配置
config.Version = sarama.V 0_11_0_0
//使用配置,新建一个异步生产者
producer, e := sarama.NewAsyncProducer ([]string{"IP:9092" ,"IP:9092" ,"IP:9092" }, config)
if e != nil {
panic(e)
}
defer producer.AsyncClose ()
//发送的消息,主题,key
msg := &sarama.ProducerMessage {
Topic: "logstash_test" ,
Key: sarama.StringEncoder ("test" ),
}
var value string
for {
value = "this is a message"
//设置发送的真正内容
fmt.Scanln (&value)
//将字符串转化为字节数组
msg.Value = sarama.ByteEncoder (value)
fmt.Println (value)
//使用通道发送
producer.Input () <- msg
//循环判断哪个通道发送过来数据.
select {
case suc := <-producer.Successes ():
fmt.Println ("offset: " , suc.Offset , "timestamp: " , suc.Timestamp .String (), "partitions: " , suc.Partition )
case fail := <-producer.Errors ():
fmt.Println ("err: " , fail.Err )
}
}
首先新建一个config
,用于配置生产者相关的配置项
通过config
和一个包含一个或多个kafka
服务器的字符串数组,新建一个producer
定义一个 生产信息 ,包括发送的主题,哪个分区,重试次数等等信息和消息内容.
通过producer
的输入通道,接受msg
如果配置中配置了,接收服务器反馈的响应,可以通过Successes
和Errors
通道来接受成功或失败的内容.
消费者
//配置
config := sarama.NewConfig ()
//接收失败通知
config.Consumer .Return .Errors = true
//设置使用的kafka版本,如果低于V0_10_0_0版本,消息中的timestrap没有作用.需要消费和生产同时配置
config.Version = sarama.V 0_11_0_0
//新建一个消费者
consumer, e := sarama.NewConsumer ([]string{"IP:9092" , "IP:9092" , "IP:9092" }, config)
if e != nil {
panic("error get consumer" )
}
defer consumer.Close ()
//根据消费者获取指定的主题分区的消费者,Offset这里指定为获取最新的消息.
partitionConsumer, err := consumer.ConsumePartition ("logstash_test" , 0 , sarama.OffsetNewest )
if err != nil {
fmt.Println ("error get partition consumer" , err)
}
defer partitionConsumer.Close ()
//循环等待接受消息.
for {
select {
//接收消息通道和错误通道的内容.
case msg := <-partitionConsumer.Messages ():
fmt.Println ("msg offset: " , msg.Offset , " partition: " , msg.Partition , " timestrap: " , msg.Timestamp .Format ("2006-Jan-02 15:04" ), " value: " , string(msg.Value ))
case err := <-partitionConsumer.Errors ():
fmt.Println (err.Err )
}
}
配置
新建一个消费者
通过消费者,指定主题的分区,获取一个特定的分区消费者.
通过分区消费者接收消息.
客户端
客户端可以用来获取消费者和生产者,还可以获取kafka的broker
信息和topic
信息,以及每个topic中的offset等.
config := sarama.NewConfig()
config.Version = sarama.V0_10_0_0
client, err := sarama.NewClient([]string {"IP:9092" , "IP:9092" , "IP:9092" }, config)
if err != nil {
panic ("client create error" )
}
defer client.Close()
topics, err := client.Topics()
if err != nil {
panic ("get topics err" )
}
for _, e := range topics {
fmt.Println(e)
}
brokers := client.Brokers()
for _, broker := range brokers {
fmt.Println(broker.Addr())
}
sarama
选项
config
结构体
实例:
config : = sarama.NewConfig()
配置包括消费者,生产者,客户端等配置,需要用到哪个指定配置哪个即可.
c.Net .MaxOpenRequests = 5
c.Net .DialTimeout = 30 * time.Second
c.Net .ReadTimeout = 30 * time.Second
c.Net .WriteTimeout = 30 * time.Second
c.Net .SASL .Handshake = true
c.Metadata .Retry .Max = 3
c.Metadata .Retry .Backoff = 250 * time.Millisecond
c.Metadata .RefreshFrequency = 10 * time.Minute
c.Metadata .Full = true
c.Producer .MaxMessageBytes = 1000000
c.Producer .RequiredAcks = WaitForLocal
c.Producer .Timeout = 10 * time.Second
c.Producer .Partitioner = NewHashPartitioner //选择分区的分区选择器.用于选择主题的分区
c.Producer .Retry .Max = 3 //重试次数
c.Producer .Retry .Backoff = 100 * time.Millisecond
c.Producer .Return .Errors = true //是否接收返回的错误消息,当发生错误时会放到Error这个通道中.从它里面获取错误消息
//抓取数据的大小设置
c.Consumer .Fetch .Min = 1
c.Consumer .Fetch .Default = 32768
c.Consumer .Retry .Backoff = 2 * time.Second //失败后再次尝试的间隔时间
c.Consumer .MaxWaitTime = 250 * time.Millisecond //最大等待时间
c.Consumer .MaxProcessingTime = 100 * time.Millisecond
c.Consumer .Return .Errors = false //是否接收返回的错误消息,当发生错误时会放到Error这个通道中.从它里面获取错误消息
c.Consumer .Offsets .CommitInterval = 1 * time.Second // 提交跟新Offset的频率
c.Consumer .Offsets .Initial = OffsetNewest // 指定Offset,也就是从哪里获取消息,默认时从主题的开始获取.
c.ClientID = defaultClientID
c.ChannelBufferSize = 256 //通道缓存大小
c.Version = minVersion //指定kafka版本,不指定,使用最小版本,高版本的新功能可能无法正常使用.
c.MetricRegistry = metrics.NewRegistry ()
生产者的分区的分割器
分区选择在多个分区存在的情况下,决定将消息发送到哪个分区.
sarama
有多个分割器:
sarama.NewManualPartitioner()
sarama.NewRandomPartitioner()
sarama.NewRoundRobinPartitioner()
sarama.NewHashPartitioner()
生产者的消息ProducerMessage
Topic string
Key Encoder
Value Encoder
Headers []RecordHeader
Metadata interface { }
Offset int64
Partition int32
Timestamp time.Time
retries int
flags flagSet
消费者信息ConsumerMessage
// ConsumerMessage encapsulates a Kafka message returned by the consumer.
type ConsumerMessage struct {
Key, Value []byte //key和保存的值
Topic string //要消费的主题
Partition int32 //要消费的分区
Offset int64 //要消费的消息的位置,从哪里开始消费,最开始的,还是最后的
Timestamp time.Time // only set if kafka is version 0.10 +, 内部时间
BlockTimestamp time .Time // only set if kafka is version 0.10 +, outer (compressed) block timestamp
Headers []*RecordHeader // only set if kafka is version 0.11 +
}