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 并将其发送出去
有疑问加站长微信联系（非本文作者）
