标签：... 入门 err Sarama broker Kafka 发送 client response

摘要

在这一篇的文章中，我将从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：如果有其他的问题，也可以在公众号找到我，欢迎来找我玩~

标签：...,入门,err,Sarama,broker,Kafka,发送,client,response
来源： https://www.cnblogs.com/hongjijun/p/13584373.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。