摘要
在这一篇的文章中，我将从Sarama的同步生产者和异步生产者怎么创建开始讲起，然后我将向你介绍生产者中的各个参数是什么，怎么使用。
然后我将从创建生产者的代码开始，按照代码的调用流程慢慢深入，直到发送消息并接收到响应。
这个过程跟上面的文章说到的kafka各个层次其实是有对应关系的。
1.如何使用
1.1 介绍
在学习如何使用Sarama生产消息之前，我先稍微介绍一下。
Sarama有两种类型的生产者，同步生产者和异步生产者。
官方文档的大致意思是异步生产者使用channel接收（生产成功或失败）的消息，并且也通过channel来发送消息，这样做通常是性能最高的。而同步生产者需要阻塞，直到收到了acks。但是这也带来了两个问题，一是性能变得更差了，而是可靠性是依靠参数acks来保证的。
1.2 异步发送
然后我们直接来看看Sarama是怎么发送异步消息的。
我们先来创建一个最简陋的异步生产者，省略所有的不必要的配置。
注意，为了更容易阅读，我删去了错误处理，并且用省略号替代。
func main() {
config := sarama.NewConfig()
client, err := sarama.NewClient([]string{"localhost:9092"}, config)
...
producer, err := sarama.NewAsyncProducerFromClient(client)
...
defer producer.Close()
topic := "topic-test"
for i := 0; i <= 100; i++ {
text := fmt.Sprintf("message %08d", i)
producer.Input() <- &sarama.ProducerMessage{
Topic: topic,
Key: nil,
Value: sarama.StringEncoder(text)}
}
}
可以看出，Sarama发送消息的套路就是先创建一个config，这里更多的config内容我们会在后文提到。
随后根据这个config，和broker地址，创建出生产者客户端。
再然后根据客户端来创建生产者对象（其实在这里用对象不够严谨，但是我认为这么理解是没有问题的）。
最后就可以使用这个生产者对象来发送信息了。
消息的构造过程中我也省略了其他的参数，只保留了最重要也是最必须的两个参数：主题和消息内容。
到了这里，一个简单的异步生产者发送消息的过程就结束了。
1.3 同步发送
在看完了异步发送之后，你可能会有很多的诸如“为什么要这么做”的疑问。
我们先来看看同步发送，再来对比一下：
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
client, err := sarama.NewClient([]string{"localhost:9092"}, config)
...
producer, err := sarama.NewSyncProducerFromClient(client)
...
defer producer.Close()
topic := "topic-test"
for i := 0; i <= 10; i++ {
text := fmt.Sprintf("message %08d", i)
partition, offset, err := producer.SendMessage(
&sarama.ProducerMessage{
Topic: topic,
Key: nil,
Value: sarama.StringEncoder(text)})
...
log.Println("send message success, partition = ", partition, " offset = ", offset)
}
}
可以看出同步发送跟异步发送的过程是很相似的。
不同的地方在于，同步生产者发送消息，使用的不是channel，并且SendMessage方法有三个返回的值，分别为这条消息的被发送到了哪个partition，处于哪个offset，是否有error。
也就是说，只有在消息成功的发送并写入了broker，才会有返回值。
2. 配置
2.1 默认配置
我们顺着源码看一下这一行：
config := sarama.NewConfig()
可以看到Sarama已经返回了一个默认的config了：
// NewConfig returns a new configuration instance with sane defaults.
func NewConfig() *Config {
c := &Config{}
c.Producer.MaxMessageBytes = 1000000
c.Producer.RequiredAcks = WaitForLocal
c.Producer.Timeout = 10 * time.Second
...
}
2.2 可选配置
我们来看看Config这个结构体，里面有哪些配置项是允许用户自定义的。
因为实在是太长了，限于篇幅以及作者的学识，在这篇文章中不能一一讲解，所以在这篇文章只会选取部分生产者相关的配置进行讲解。
但是无论是Golang客户端，还是Java客户端，都不重要，你只需要知道哪些参数对于你的生产者的生产速度、消息的可靠性等有关系就可以了。
// Config is used to pass multiple configuration options to Sarama's constructors.
type Config struct {
Admin struct {
...
}
Net struct {
...
}
Metadata struct {
...
}
Producer struct {
...
}
Consumer struct {
...
}
ClientID string
...
}
我们可以看出，关于Sarama的配置，分成了很多个部分，我们来具体看一看Producer的这部分。
2.3 重要的生产者参数
在这里我打算介绍一部分我个人认为比较重要的生产者参数。
MaxMessageBytes int
这个参数影响了一条消息的最大字节数，默认是1000000。但是注意，这个参数必须要小于broker中的 message.max.bytes。
RequiredAcks RequiredAcks
这个参数影响了消息需要被多少broker写入之后才返回。取值可以是0、1、-1，分别代表了不需要等待broker确认才返回、需要分区的leader确认后才返回、以及需要分区的所有副本确认后返回。
Partitioner PartitionerConstructor
这个是分区器。Sarama默认提供了几种分区器，如果不指定默认使用Hash分区器。
Retry
这个参数代表了重试的次数，以及重试的时间，主要发生在一些可重试的错误中。
Flush
用于设置将消息打包发送，简单来讲就是每次发送消息到broker的时候，不是生产一条消息就发送一条消息，而是等消息累积到一定的程度了，再打包发送。所以里面含有两个参数。一个是多少条消息触发打包发送，一个是累计的消息大小到了多少，然后发送。
2.4 幂等生产者
在聊幂等生产者之前，我们先来看看生产者中另外一个很重要的参数：
MaxOpenRequests int
这个参数代表了允许没有收到acks而可以同时发送的最大batch数。
Idempotent bool
用于幂等生产者，当这一项设置为true的时候，生产者将保证生产的消息一定是有序且精确一次的。
为什么会需要这个选项呢？
当MaxOpenRequests这个参数配置大于1的时候，代表了允许有多个请求发送了还没有收到回应。假设此时的重试次数也设置为了大于1，当同时发送了2个请求，如果第一个请求发送到broker中，broker写入失败了，但是第二个请求写入成功了，那么客户端将重新发送第一个消息的请求，这个时候会造成乱序。
又比如当第一个请求返回acks的时候，因为网络原因，客户端没有收到，所以客户端进行了重发，这个时候就会造成消息的重复。
所以，幂等生产者就是为了保证消息发送到broker中是有序且不重复的。
消息的有序可以通过MaxOpenRequests设置为1来保证，这个时候每个消息必须收到了acks才能发送下一条，所以一定是有序的，但是不能够保证不重复。
而且当MaxOpenRequests设置为1的时候，吞吐量不高。
注意，当启动幂等生产者的时候，Retry次数必须要大于0，ack必须为all。
在Java客户端中，允许MaxOpenRequests小于等于5。
但是在Sarama中有一个很奇怪的地方我也没有研究明白，我们直接看一看这部分的代码：
if c.Producer.Idempotent {
if !c.Version.IsAtLeast(V0_11_0_0) {
return ConfigurationError("Idempotent producer requires Version >= V0_11_0_0")
}
if c.Producer.Retry.Max == 0 {
return ConfigurationError("Idempotent producer requires Producer.Retry.Max >= 1")
}
if c.Producer.RequiredAcks != WaitForAll {
return ConfigurationError("Idempotent producer requires Producer.RequiredAcks to be WaitForAll")
}
if c.Net.MaxOpenRequests > 1 {
return ConfigurationError("Idempotent producer requires Net.MaxOpenRequests to be 1")
}
}
这一部分第一项是版本号，没问题，第二第三项是Retry和Acks，也没有问题。问题在于第四项，这里的MaxOpenRequests参数，我想应该等同于Java客户端中的MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION，按照Java客户端中的配置，应该是这个参数小于等于5，即可保证幂等，但是这里必须得设置为1。
检查了Sarama的Issue，有开发者提出了这个问题，但是目前作者还没有打算解决。
3 broker
在这一节的内容中，我将会从代码的层面介绍 Sarama 生产者发送消息的全过程。但是因为代码很多，我将会省略一些内容，包括一些错误处理、重试等。
这些都很重要，也不应该被省略。但是因为篇幅有限，我只能介绍最核心的发送消息这一部分的内容。
我会在贴代码之前，大概的说一下这段代码的思路。随后，我会在代码中加入一些注释，来更详细的进行解释。
然后我们开始吧！
producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)
一切都从这么一行开始讲起。
我们进去看看。
在这里其实就只有两个部分，先是通过地址和配置，构建一个 client 。
func NewAsyncProducer(addrs []string, conf *Config) (AsyncProducer, error) {
// 构建client
client, err := NewClient(addrs, conf)
if err != nil {
return nil, err
}
// 构建AsyncProducer
return newAsyncProducer(client)
}
3.1 Client的创建
在创建 Client 的过程中，先构建一个 client 结构体。
里面的参数我们先不管，等用到了再进行解释。
然后创建完之后，刷新元数据，并且启动一个协程，在后台进行刷新。
func NewClient(addrs []string, conf *Config) (Client, error) {
...
// 构建一个client
client := &client{
conf: conf,
closer: make(chan none),
closed: make(chan none),
brokers: make(map[int32]*Broker),
metadata: make(map[string]map[int32]*PartitionMetadata),
metadataTopics: make(map[string]none),
cachedPartitionsResults: make(map[string][maxPartitionIndex][]int32),
coordinators: make(map[string]int32),
}
// 把用户输入的broker地址作为“种子broker”增加到seedBrokers中
// 随后客户端会根据已有的broker地址，自动刷新元数据，以获取更多的broker地址
// 所以称之为种子
random := rand.New(rand.NewSource(time.Now().UnixNano()))
for _, index := range random.Perm(len(addrs)) {
client.seedBrokers = append(client.seedBrokers, NewBroker(addrs[index]))
}
...
// 启动协程在后台刷新元数据
go withRecover(client.backgroundMetadataUpdater)
return client, nil
}
3.2 元数据的更新
后台更新元数据的设计其实很简单，利用一个 ticker ，按时对元数据进行更新，直到 client 关闭。
这里先提一下我们说的元数据，有哪些内容。
你可以简单的理解为包含了所有 broker 的地址（因为 broker 可能新增，也可能减少），以及包含了哪些 topic ，这些 topic 有哪些 partition 等。
func (client *client) backgroundMetadataUpdater() {
// 按照配置的时间更新元数据
ticker := time.NewTicker(client.conf.Metadata.RefreshFrequency)
defer ticker.Stop()
// 循环获取channel，判断是执行更新操作还是终止
for {
select {
case <-ticker.C:
if err := client.refreshMetadata(); err != nil {
Logger.Println("Client background metadata update:", err)
}
case <-client.closer:
return
}
}
}
然后我们继续来看看 client.refreshMetadata() 这个方法，这个方法是判断了一下需要刷新哪些主题的元数据，还是说全部主题的元数据。
然后我们继续。
在这里也还没有涉及到具体的更新操作。我们看 tryRefreshMetadata 这个方法的参数可以得知，在这里我们设置了需要刷新元数据的主题，重试的次数，超时的时间。
func (client *client) RefreshMetadata(topics ...string) error {
deadline := time.Time{}
if client.conf.Metadata.Timeout > 0 {
deadline = time.Now().Add(client.conf.Metadata.Timeout)
}
// 设置参数
return client.tryRefreshMetadata(topics, client.conf.Metadata.Retry.Max, deadline)
}
然后终于来到了tryRefreshMetadata这个方法。
在这个方法中，会选取已经存在的broker，构造获取元数据的请求。
在收到回应后，如果不存在任何的错误，就将这些元数据用于更新客户端。
func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int, deadline time.Time) error {
...
broker := client.any()
for ; broker != nil && !pastDeadline(0); broker = client.any() {
...
req := &MetadataRequest{
Topics: topics,
// 是否允许创建不存在的主题
AllowAutoTopicCreation: allowAutoTopicCreation
}
response, err := broker.GetMetadata(req)
switch err.(type) {
case nil:
allKnownMetaData := len(topics) == 0
// 对元数据进行更新
shouldRetry, err := client.updateMetadata(response, allKnownMetaData)
if shouldRetry {
Logger.Println("client/metadata found some partitions to be leaderless")
return retry(err)
}
return err
case ...
...
}
}
然后我们继续往下看看当客户端拿到了 response 之后，是如何更新的。
首先，先对本地保存 broker 进行更新。
然后，对 topic 进行更新，以及这个 topic 下面的那些 partition 。
func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bool) (retry bool, err error) {
...
// 假设返回了新的broker id，那么保存这些新的broker，这意味着增加了broker、或者下线的broker重新上线了
// 如果返回的id我们已经保存了，但是地址变化了，那么更新地址
// 如果本地保存的一些id没有返回，说明这些broker下线了，那么删除他们
client.updateBroker(data.Brokers)
// 然后对topic也进行元数据的更新
// 主要是更新topic以及topic对应的partition
for _, topic := range data.Topics {
...
// 更新每个topic以及对应的partition
client.metadata[topic.Name] = make(map[int32]*PartitionMetadata, len(topic.Partitions))
for _, partition := range topic.Partitions {
client.metadata[topic.Name][partition.ID] = partition
...
}
}
至此，我们元数据的更新就说完了。
下面我们来说一说在更新元数据之前，broker是如何建立连接的，以及请求是如何发送出去，又是如何被broker接收的。
3.3 与Broker建立连接
让我们回到 tryRefreshMetadata 这个方法中。
这个方法里面有这么一行代码：
broker := client.any()
我们进去看看。
在这个方法里， 如果 seedBrokers 存在，那么就打开它，否则的话打开其他的broker。
注意，这里提到的其他的broker，可能是在刷新元数据的时候，获取到的。这就跟上面的内容联系在一起了。
func (client *client) any() *Broker {
...
if len(client.seedBrokers) > 0 {
_ = client.seedBrokers[0].Open(client.conf)
return client.seedBrokers[0]
}
// 不保证一定是按顺序的
for _, broker := range client.brokers {
_ = broker.Open(client.conf)
return broker
}
return nil
}
然后再让我们看看 Open方法做了什么。
Open方法异步的建立了一个tcp连接，然后创建了一个缓冲大小为MaxOpenRequests的channel。
这个名为 responses 的 channel ，用于接收从 broker发送回来的消息。
其实在 broker 中，用于发送消息跟接收消息的 channel 都设置成了这个大小。
MaxOpenRequests 这个参数你可以理解为是Java客户端中的max.in.flight.requests.per.connection。
然后，又启动了一个协程，用于接收消息。
func (b *Broker) Open(conf *Config) error {
if conf == nil {
conf = NewConfig()
}
...
go withRecover(func() {
...
dialer := conf.getDialer()
b.conn, b.connErr = dialer.Dial("tcp", b.addr)
...
b.responses = make(chan responsePromise, b.conf.Net.MaxOpenRequests-1)
...
go withRecover(b.responseReceiver)
})
3.4 从Broker接收响应
我们来看看 responseReceiver 是怎么工作的。
其实很容易理解，当 broker 收到一个 response 的时候，先解析消息的头部，然后再解析消息的内容。并把这些内容写进 response 的 packets 中。
func (b *Broker) responseReceiver() {
for response := range b.responses {
...
// 先根据Header的版本读取对应长度的Header
var headerLength = getHeaderLength(response.headerVersion)
header := make([]byte, headerLength)
bytesReadHeader, err := b.readFull(header)
decodedHeader := responseHeader{}
err = versionedDecode(header, &decodedHeader, response.headerVersion)
...
// 解析具体的内容
buf := make([]byte, decodedHeader.length-int32(headerLength)+4)
bytesReadBody, err := b.readFull(buf)
// 省略了一些错误处理，总之，如果发生了错误，就把错误信息写进 response.errors 中
response.packets <- buf
}
}
其实接收响应这部分的代码逻辑很容易理解，就是当 response 这个 channel 有了消息，就读取，然后将读取到的内容写进 response 中。
那么你可能会有一个问题，什么时候才会往response 这个 channel 发送消息呢？
很容易可以猜到，当我们发送了消息给 broker ，就应该要通知 receiver ，准备接受消息了。
既然如此，我们继续刚刚刷新元数据的部分，看看 sarama 是如何把消息发送出去的。
3.5 发送与接受消息
我们回到这一行代码：
response, err := broker.GetMetadata(req)
我们直接进去，发现在这里构造了一个接受返回信息的结构体，然后调用了sendAndReceive方法。
func (b *Broker) GetMetadata(request *MetadataRequest) (*MetadataResponse, error) {
response := new(MetadataResponse)
err := b.sendAndReceive(request, response)
if err != nil {
return nil, err
}
return response, nil
}
我们继续往下。
在这里我们可以看到，先是调用了send方法，然后返回了一个promise。并且当有消息写入这个promise的时候，就得到了结果。
而且回想一下我们在receiver中，是不是把获取到的 response 写进了 packets ，把错误结果写进了 errors 呢，跟这里是一致的对吧？
func (b *Broker) sendAndReceive(req protocolBody, res protocolBody) error {
responseHeaderVersion := int16(-1)
if res != nil {
responseHeaderVersion = res.headerVersion()
}
promise, err := b.send(req, res != nil, responseHeaderVersion)
if err != nil {
return err
}
if promise == nil {
return nil
}
// 这里的promise，是上面send方法返回的
select {
case buf := <-promise.packets:
return versionedDecode(buf, res, req.version())
case err = <-promise.errors:
return err
}
}
带着这个想法，我们看看 send 方法做了什么事。
这个地方很重要，也是我认为 Sarama 设计的特别巧妙的一个地方。
在send方法中，把需要发送的消息通过与broker的tcp连接，同步发送到broker中。
然后构建了一个responsePromise类型的channel，然后直接将这个结构体丢进这个channel中。然后回想一下，我们在responseReceiver这个方法中，不断消费接收到的response。
此时在responseReceiver中，收到了send方法传递的responsePromise，他就会通过conn来读取数据，然后将数据写入这个responsePromise的packets中，或者将错误信息写入errors中。
而此时，再看看send方法，他返回了这个responsePromise的指针。所以，sendAndReceive方法就在等待这个responsePromise内的packets或者errors的channel被写入数据。当responseReceiver接收到了响应并且写入数据的时候，packets或者errors就会被写入消息。
func (b *Broker) send(rb protocolBody, promiseResponse bool, responseHeaderVersion int16) (*responsePromise, error) {
...
// 将请求的内容封装进 request ，然后发送到Broker中
// 注意一下这里的 b.write(buf)
// 里面做了 b.conn.Write(buf) 这件事情
req := &request{correlationID: b.correlationID, clientID: b.conf.ClientID, body: rb}
buf, err := encode(req, b.conf.MetricRegistry)
bytes, err := b.write(buf)
...
// 如果我们的response为nil，也就是说当不需要response的时候，是不会放进inflight发送队列的
if !promiseResponse {
// Record request latency without the response
b.updateRequestLatencyAndInFlightMetrics(time.Since(requestTime))
return nil, nil
}
// 构建一个接收响应的 channel ，返回这个channel的指针
// 这个 channel 内部包含了两个 channel，一个用来接收响应，一个用来接收错误
promise := responsePromise{requestTime, req.correlationID, responseHeaderVersion, make(chan []byte), make(chan error)}
b.responses <- promise
// 这里返回指针特别的关键，是把消息的发送跟消息的接收联系在一起了
return &promise, nil
}
让我们来用一张图说明一下上面这个发送跟接收的过程：
这一段比较绕，但这也是Sarama发送与接受消息的核心内容，希望我的解释能够让你理解：）
4 AsyncProcuder
在上一节中，我们已经分析了client的构造全过程，并且在构造client刷新元数据的时候，也解释了sarama是如何发送消息以及接受消息的。
在这一节中，我打算解释一下AsyncProcuder是如何发送消息的。
因为有了上一节的铺垫，这一节的内容应该会比较容易理解。
我们从newAsyncProducer(client)这一行开始讲起。
我们先说说input:make(chan *ProducerMessage)，这个事关我们的消息发送。注意到这个channel是没有缓冲的。
也就是说当我们发送一条消息到input中的时候，此时发送方会阻塞，这说明了之后的操作必须不能够被阻塞，否则会影响消息的发送效率。
然后其他字段我们先不管，后面用到了我们再提。
func newAsyncProducer(client Client) (AsyncProducer, error) {
...
p := &asyncProducer{
client: client,
conf: client.Config(),
errors: make(chan *ProducerError),
input: make(chan *ProducerMessage),
successes: make(chan *ProducerMessage),
retries: make(chan *ProducerMessage),
brokers: make(map[*Broker]*brokerProducer),
brokerRefs: make(map[*brokerProducer]int),
txnmgr: txnmgr,
}
go withRecover(p.dispatcher)
go withRecover(p.retryHandler)
}
4.1 dispatcher
我们往下看看下面协程启动的go withRecover(p.dispatcher)。
在这个方法中，首先创建了一个以Topic为key的map，这个map的value是无缓冲的channel。
到这里我们很容易可以推测得出，当通过input发送一条消息的时候，消息会到dispatcher这里，被分配到各个Topic中。
注意，在这个时候，channel还是无缓冲的，所以我们可以推测下一步的操作，依旧是无阻塞的。
func (p *asyncProducer) dispatcher() {
handlers := make(map[string]chan<- *ProducerMessage)
...
for msg := range p.input {
...
// 拦截器
for _, interceptor := range p.conf.Producer.Interceptors {
msg.safelyApplyInterceptor(interceptor)
}
...
// 找到这个Topic对应的Handler
handler := handlers[msg.Topic]
if handler == nil {
// 如果此时还不存在这个Topic对应的Handler，那么创建一个
// 虽然说他叫Handler，但他其实是一个无缓冲的
handler = p.newTopicProducer(msg.Topic)
handlers[msg.Topic] = handler
}
// 然后把这条消息写进这个Handler中
handler <- msg
}
}
然后让我们来handler = p.newTopicProducer(msg.Topic)这一行的代码。
在这里创建了一个缓冲大小为ChannelBufferSize的channel，用于存放发送到这个主题的消息。
然后创建了一个topicProducer，在这个时候你可以认为消息已经交付给各个topic的topicProducer了。
func (p *asyncProducer) newTopicProducer(topic string) chan<- *ProducerMessage {
input := make(chan *ProducerMessage, p.conf.ChannelBufferSize)
tp := &topicProducer{
parent: p,
topic: topic,
input: input,
breaker: breaker.New(3, 1, 10*time.Second),
handlers: make(map[int32]chan<- *ProducerMessage),
partitioner: p.conf.Producer.Partitioner(topic),
}
go withRecover(tp.dispatch)
return input
}
4.2 topicDispatch
然后我们来看看go withRecover(tp.dispatch)这一行代码。
同样是启动了一个协程，来处理消息。
也就是说，到了这一步，对于每一个Topic，都有一个协程来处理消息。
在这个dispatch()方法中，也同样的接收到一条消息，就会去找这条消息所在的分区的channel，然后把消息写进去。
func (tp *topicProducer) dispatch() {
for msg := range tp.input {
...
// 同样是找到这条消息所在的分区对应的channel，然后把消息丢进去
handler := tp.handlers[msg.Partition]
if handler == nil {
handler = tp.parent.newPartitionProducer(msg.Topic, msg.Partition)
tp.handlers[msg.Partition] = handler
}
handler <- msg
}
}
4.3 PartitionDispatch
我们进tp.parent.newPartitionProducer(msg.Topic, msg.Partition)这里看看。
你可以发现partitionProducer跟topicProducer是很像的。
其实他们就是代表了一条消息的分发，从producer到topic到partition。
注意，这里面的channel缓冲大小，也是ChannelBufferSize。
func (p *asyncProducer) newPartitionProducer(topic string, partition int32) chan<- *ProducerMessage {
input := make(chan *ProducerMessage, p.conf.ChannelBufferSize)
pp := &partitionProducer{
parent: p,
topic: topic,
partition: partition,
input: input,
breaker: breaker.New(3, 1, 10*time.Second),
retryState: make([]partitionRetryState, p.conf.Producer.Retry.Max+1),
}
go withRecover(pp.dispatch)
return input
}
4.4 partitionProducer
到了这一步，我们再来看看消息到了每个partition所在的channel，是如何处理的。
其实在这一步中，主要是做一些错误处理之类的，然后把消息丢进brokerProducer。
可以理解为这一步是业务逻辑层到网络IO层的转变，在这之前我们只关心消息去到了哪个分区，而在这之后，我们需要找到这个分区所在的broker的地址，并使用之前已经建立好的TCP连接，发送这条消息。
func (pp *partitionProducer) dispatch() {
// 找到这个主题和分区的leader所在的broker
pp.leader, _ = pp.parent.client.Leader(pp.topic, pp.partition)
// 如果此时找到了这个leader
if pp.leader != nil {
pp.brokerProducer = pp.parent.getBrokerProducer(pp.leader)
pp.parent.inFlight.Add(1)
// 发送一条消息来表示同步
pp.brokerProducer.input <- &ProducerMessage{Topic: pp.topic, Partition: pp.partition, flags: syn}
}
...// 各种异常情况
// 然后把消息丢进brokerProducer中
pp.brokerProducer.input <- msg
}
4.5 brokerProducer
到了这里，大概算是整个发送流程最后的一个步骤了。
我们来看看pp.parent.getBrokerProducer(pp.leader)这行代码里面的内容。
其实就是找到asyncProducer中的brokerProducer，如果不存在，则创建一个。
func (p *asyncProducer) getBrokerProducer(broker *Broker) *brokerProducer {
p.brokerLock.Lock()
defer p.brokerLock.Unlock()
bp := p.brokers[broker]
if bp == nil {
bp = p.newBrokerProducer(broker)
p.brokers[broker] = bp
p.brokerRefs[bp] = 0
}
p.brokerRefs[bp]++
return bp
}
那我们就来看看brokerProducer是怎么创建出来的。
看这个方法中启动的第二个协程，我们可以推测bridge这个channel收到消息后，会把收到的消息打包成一个request，然后调用Produce方法。
并且，将返回的结果的指针地址，写进response中。
然后构造好brokerProducerResponse，并且写入responses中。
func (p *asyncProducer) newBrokerProducer(broker *Broker) *brokerProducer {
var (
input = make(chan *ProducerMessage)
bridge = make(chan *produceSet)
responses = make(chan *brokerProducerResponse)
)
bp := &brokerProducer{
parent: p,
broker: broker,
input: input,
output: bridge,
responses: responses,
stopchan: make(chan struct{}),
buffer: newProduceSet(p),
currentRetries: make(map[string]map[int32]error),
}
go withRecover(bp.run)
// minimal bridge to make the network response `select`able
go withRecover(func() {
for set := range bridge {
request := set.buildRequest()
response, err := broker.Produce(request)
responses <- &brokerProducerResponse{
set: set,
err: err,
res: response,
}
}
close(responses)
})
if p.conf.Producer.Retry.Max <= 0 {
bp.abandoned = make(chan struct{})
}
return bp
}
让我们再来看看broker.Produce(request)这一行代码。
是不是很熟悉呢，我们在client部分讲到的sendAndReceive方法。
而且我们可以发现，如果我们设置了需要Acks，就会返回一个response；如果没设置，那么消息发出去之后，就不管了。
此时在获取了response，并且填入了response的内容后，返回这个response的内容。
func (b *Broker) Produce(request *ProduceRequest) (*ProduceResponse, error) {
var (
response *ProduceResponse
err error
)
if request.RequiredAcks == NoResponse {
err = b.sendAndReceive(request, nil)
} else {
response = new(ProduceResponse)
err = b.sendAndReceive(request, response)
}
if err != nil {
return nil, err
}
return response, nil
}
至此，Sarama生产者相关的内容就介绍完毕了。
写在最后
这一篇写的实在是有些久了。
主要是作者这段时间实在是太忙了，还没有完全平衡好目前的学习工作和生活，导致每天花在学习上的时间不多，效率也不高。
另外就是网上我也没有查到有Sarama相关的解析，都是一些API的调用。因为作者恰好开始学习Kafka，为了更好地了解生产者的每一个参数，我选择去研究生产者客户端。
但是，因为作者源码阅读能力实在是有限，在这个过程中很有可能会有一些错误的理解。所以当你发现了一些违和的地方，也请不吝指教，谢谢你！
再次感谢你能看到这里！
PS：如果有其他的问题，也可以在公众号找到我，欢迎来找我玩~