Kafka入门（3）：Sarama生产者是如何工作的

摘要

在这一篇的文章中，我将从Sarama的同步生产者和异步生产者怎么创建开始讲起，然后我将向你介绍生产者中的各个参数是什么，怎么使用。

然后我将从创建生产者的代码开始，按照代码的调用流程慢慢深入，直到发送消息并接收到响应。

这个过程跟上面的文章说到的kafka各个层次其实是有对应关系的。

1.如何使用

1.1 介绍

在学习如何使用Sarama生产消息之前，我先稍微介绍一下。

Sarama有两种类型的生产者，同步生产者和异步生产者。

To produce messages, use either the AsyncProducer or the SyncProducer. The AsyncProducer accepts messages on a channel and produces them asynchronously in the background as efficiently as possible; it is preferred in most cases. The SyncProducer provides a method which will block until Kafka acknowledges the message as produced. This can be useful but comes with two caveats: it will generally be less efficient, and the actual durability guarantees depend on the configured value of Producer.RequiredAcks. There are configurations where a message acknowledged by the SyncProducer can still sometimes be lost.

官方文档的大致意思是异步生产者使用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：如果有其他的问题，也可以在公众号找到我，欢迎来找我玩~