kafka golang 客户端sarama 生产者代码解析

1. syncProducer 和asyncProducer的关系

syncProducer 是所有功能都是由asyncProducer实现的,而syncProducer 之所以可以同步发送消息,答案就在SendMessage 函数中,源码如下

 func(sp *syncProducer)SendMessage(msg *ProducerMessage) (partitionint32,offsetint64,errerror) {
 expectation :=make(chan*ProducerError,1)
 msg.expectation = expectation
 sp.producer.Input() <- msg
 if err := <-expectation;err != nil { // 阻塞等待返回结果
 return-1,-1,err.Err
 }
 return msg.Partition,msg.Offset,nil
}

而使用asyncProducer 时,只需要 直接将信息producer.Input()<-&ProducerMessage{} 放入进producer.Input(), 然后异步读取返回结果 chan*ProducerError

2. 消息传递过程

 // one per topic
// partitions messages, then dispatches them by partition
type topicProducer struct{
 parent *asyncProducer
 topic string
 input <-chan*ProducerMessage
 breaker *breaker.Breaker
 handlers map[int32] chan<- *ProducerMessage
 partitioner Partitioner
}
type brokerProducer struct{
 parent *asyncProducer
 broker *Broker
 input <-chan*ProducerMessage
 output chan<- *produceSet
 responses <-chan*brokerProducerResponse
 buffer *produceSet
 timer <-chantime.Time
 timerFired bool
 closing error
 currentRetries map[string]map[int32]error
}

由代码可以看出topicProducer,partitionProducer,brokerProducer的parent都是asyncProducer

消息传递过程:
asyncProducer.dispatcher ->topicProducer.dispath -> partitionProducer.dispatch -> brokerProducer ->produceSet

其中produceSet 对消息进行聚集,若配置了压缩的参数,则会压缩一个set中的所有的msg, 即批量压缩, 然后构建一个ProduceRequest ,然后由 broker.Produce 将请求发送出去,其中 broker 结构体代表一个kafka broker 的连接
partitionProducer 会选择leader broker地址 ,若选择失败,则会重新选择leader broker ,然后由这个连接发送消息

根据kafka版本不同,消息会放入到不同的结构体中若版本大于V0.11,set.recordsToSend.RecordBatch.addRecord(rec) 将一个rec添加进去,否则将set.recordsToSend.MsgSet.addMessage(msgToSend)

在生成一个newBrokerProducer时,broker会开启消费output, 而output就是一个存放produceSet的channel,阻塞等待刷新ProduceRequest 并将其发送出去

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传