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 并将其发送出去