sarama源码分析Part I

//GetMetadata send a metadata request and returns a metadata response or error
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
}

  func (b *Broker) sendAndReceive(req protocolBody, res versionedDecoder) error {
      // send 返回一个 promise
      promise, err := b.send(req, res != nil)
      if err != nil {
          return err
      }

      // 当配置`RequiredAcks`=0(即不关注server返回)时，返回的promise会是nil
      if promise == nil {
          return nil
      }

      // 然后就等待这个 promise
      select {
      case buf := <-promise.packets:
          return versionedDecode(buf, res, req.version())
      case err = <-promise.errors:
          return err
      }
  }

// 为节省篇幅，省略了一些错误处理的展示
  func (b *Broker) send(rb protocolBody, promiseResponse bool) (*responsePromise, error) {
      b.lock.Lock()
      defer b.lock.Unlock()

....
      // request的correlationID
      req := &request{correlationID: b.correlationID, clientID: b.conf.ClientID, body: rb}
      buf, err := encode(req, b.conf.MetricRegistry)
      if err != nil {
          return nil, err
      }

      requestTime := time.Now()
      // Will be decremented in responseReceiver (except error or request with NoResponse)
      b.addRequestInFlightMetrics(1)
      // 关键在这里了，这里把请求发出去了
      // 最终会调用`conn.Write`
      bytes, err := b.write(buf)
      b.updateOutgoingCommunicationMetrics(bytes)
      if err != nil {
          b.addRequestInFlightMetrics(-1)
          return nil, err
      }
      b.correlationID++

      if !promiseResponse {
          // Record request latency without the response
          b.updateRequestLatencyAndInFlightMetrics(time.Since(requestTime))
          return nil, nil
      }

      // 组装了一个promise，然后把它发给了一个channel：b.responses
      // 注意这个promise里也有两个channel，它是用来返回结果的

      // 关联上了req.correlationID
      // 通过这个correlationID来关联req和response
      promise := responsePromise{requestTime, req.correlationID, make(chan []byte), make(chan error)}
      b.responses <- promise

      return &promise, nil
  }

  func (b *Broker) responseReceiver() {
      var dead error
      header := make([]byte, 8)

      // 注意是一个循环读取消息，即使没有消息
      // 也会等待在这里，除非有人关闭了这个channel
      for response := range b.responses {
          ...
          // 读header
          bytesReadHeader, err := b.readFull(header)
          requestLatency := time.Since(response.requestTime)
          ....

          decodedHeader := responseHeader{}
          err = decode(header, &decodedHeader)
          ...

          // 这里会校验收到的消息是否是所预期的，若对不上直接抛弃，继续处理下一条
          if decodedHeader.correlationID != response.correlationID {
              b.updateIncomingCommunicationMetrics(bytesReadHeader, requestLatency)
              // TODO if decoded ID < cur ID, discard until we catch up
              // TODO if decoded ID > cur ID, save it so when cur ID catches up we have a response
              dead = PacketDecodingError{fmt.Sprintf("correlation ID didn't match, wanted %d, got %d", response.correlationID, decodedHeader.correlationID)}
              response.errors <- dead
              continue
          }

          // 读body
          buf := make([]byte, decodedHeader.length-4)
          bytesReadBody, err := b.readFull(buf)
          b.updateIncomingCommunicationMetrics(bytesReadHeader+bytesReadBody, requestLatency)
          if err != nil {
              dead = err
              response.errors <- err
              continue
          }

          // 将读到的body字节通过 channel 返回
          response.packets <- buf
      }
      close(b.done)
  }

  func (b *Broker) Open(conf *Config) error {
      if !atomic.CompareAndSwapInt32(&b.opened, 0, 1) {
          return ErrAlreadyConnected
      }

      if conf == nil {
          conf = NewConfig()
      }

      err := conf.Validate()
      if err != nil {
          return err
      }

      b.lock.Lock()

      go withRecover(func() {
          defer b.lock.Unlock()

          dialer := net.Dialer{
              Timeout:   conf.Net.DialTimeout,
              KeepAlive: conf.Net.KeepAlive,
              LocalAddr: conf.Net.LocalAddr,
          }

          ...

          b.conn = newBufConn(b.conn)

          b.conf = conf

          ....

          if b.id >= 0 {
              b.registerMetrics()
          }

          ....

          b.done = make(chan bool)
          b.responses = make(chan responsePromise, b.conf.Net.MaxOpenRequests-1)

          if b.id >= 0 {
              Logger.Printf("Connected to broker at %s (registered as #%d)\n", b.addr, b.id)
          } else {
              Logger.Printf("Connected to broker at %s (unregistered)\n", b.addr)
          }
          // 这里单独起了一个协程来处理
          go withRecover(b.responseReceiver)
      })

func NewClient(addrs []string, conf *Config) (Client, error) {
    Logger.Println("Initializing new client")
    ...
    if conf.Metadata.Full {
        // do an initial fetch of all cluster metadata by specifying an empty list of topics
        // 这里会从broker拉取一次metadata
        // 在此过程中首先会先open一个broker
        err := client.RefreshMetadata()
        switch err {
        case nil:
            break
        case ErrLeaderNotAvailable, ErrReplicaNotAvailable, ErrTopicAuthorizationFailed, ErrClusterAuthorizationFailed:
            // indicates that maybe part of the cluster is down, but is not fatal to creating the client
            Logger.Println(err)
        default:
            close(client.closed) // we haven't started the background updater yet, so we have to do this manually
            _ = client.Close()
            return nil, err
        }
    }
    // 这里启动后台定时拉取metadata的任务
    go withRecover(client.backgroundMetadataUpdater)

    Logger.Println("Successfully initialized new client")

    return client, nil
    }

// one per broker; also constructs an associated flusher
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
    // 注意这里是按set来发的，消息走到这里时已经被整合成一个batch了
    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
}

// 此处逻辑位于 func (bp *brokerProducer) handleSuccess 方法中
        sent.eachPartition(func(topic string, partition int32, pSet *partitionSet) {
            block := response.GetBlock(topic, partition)
            if block == nil {
                // handled in the previous "eachPartition" loop
                return
            }

            switch block.Err {
            case ErrInvalidMessage, ErrUnknownTopicOrPartition, ErrLeaderNotAvailable, ErrNotLeaderForPartition,
                ErrRequestTimedOut, ErrNotEnoughReplicas, ErrNotEnoughReplicasAfterAppend:
                Logger.Printf("producer/broker/%d state change to [retrying] on %s/%d because %v\n",
                    bp.broker.ID(), topic, partition, block.Err)
                if bp.currentRetries[topic] == nil {
                    bp.currentRetries[topic] = make(map[int32]error)
                }
                // 这里是所谓的`sets a flag for the given topic/partition`
                // 用于标识这个topic/partition是有问题的
                bp.currentRetries[topic][partition] = block.Err
                // 以下为将消息发送至 retryHandler
                if bp.parent.conf.Producer.Idempotent {
                    go bp.parent.retryBatch(topic, partition, pSet, block.Err)
                } else {
                    bp.parent.retryMessages(pSet.msgs, block.Err)
                }
                // dropping the following messages has the side effect of incrementing their retry count
                bp.parent.retryMessages(bp.buffer.dropPartition(topic, partition), block.Err)
            }
        })

// 此处逻辑位于 func (bp *brokerProducer) run 方法
            if reason := bp.needsRetry(msg); reason != nil {
                bp.parent.retryMessage(msg, reason)

                if bp.closing == nil && msg.flags&fin == fin {
                    // we were retrying this partition but we can start processing again
                    delete(bp.currentRetries[msg.Topic], msg.Partition)
                    Logger.Printf("producer/broker/%d state change to [closed] on %s/%d\n",
                        bp.broker.ID(), msg.Topic, msg.Partition)
                }

                continue
            }

func (bp *brokerProducer) needsRetry(msg *ProducerMessage) error {
    if bp.closing != nil {
        return bp.closing
    }

    return bp.currentRetries[msg.Topic][msg.Partition]
}

// 此处逻辑位于 func (pp *partitionProducer) dispatch 中
        if msg.retries > pp.highWatermark {
            // a new, higher, retry level; handle it and then back off
            pp.newHighWatermark(msg.retries)
            pp.backoff(msg.retries)
        }

func (pp *partitionProducer) newHighWatermark(hwm int) {
    Logger.Printf("producer/leader/%s/%d state change to [retrying-%d]\n", pp.topic, pp.partition, hwm)
    pp.highWatermark = hwm

    // send off a fin so that we know when everything "in between" has made it
    // back to us and we can safely flush the backlog (otherwise we risk re-ordering messages)
    pp.retryState[pp.highWatermark].expectChaser = true
    pp.parent.inFlight.Add(1) // we're generating a fin message; track it so we don't shut down while it's still inflight
    // 注意此处的retries做了减一操作
    // 这样的话，当brokerProducer收到这个fin消息，并重新retry到partitionProducer中时
    // 才会满足msg.retries == highWatermark 从而被识别
    pp.brokerProducer.input <- &ProducerMessage{Topic: pp.topic, Partition: pp.partition, flags: fin, retries: pp.highWatermark - 1}

    // a new HWM means that our current broker selection is out of date
    Logger.Printf("producer/leader/%s/%d abandoning broker %d\n", pp.topic, pp.partition, pp.leader.ID())
    pp.parent.unrefBrokerProducer(pp.leader, pp.brokerProducer)
    pp.brokerProducer = nil
}

// 依然在 func (pp *partitionProducer) dispatch 中
        // if we made it this far then the current msg contains real data, and can be sent to the next goroutine
        // without breaking any of our ordering guarantees

        if pp.brokerProducer == nil {
            if err := pp.updateLeader(); err != nil {
                pp.parent.returnError(msg, err)
                pp.backoff(msg.retries)
                continue
            }
            Logger.Printf("producer/leader/%s/%d selected broker %d\n", pp.topic, pp.partition, pp.leader.ID())
        }

        // Now that we know we have a broker to actually try and send this message to, generate the sequence
        // number for it.
        // All messages being retried (sent or not) have already had their retry count updated
        // Also, ignore "special" syn/fin messages used to sync the brokerProducer and the topicProducer.
        if pp.parent.conf.Producer.Idempotent && msg.retries == 0 && msg.flags == 0 {
            msg.sequenceNumber, msg.producerEpoch = pp.parent.txnmgr.getAndIncrementSequenceNumber(msg.Topic, msg.Partition)
            msg.hasSequence = true
        }

        pp.brokerProducer.input <- msg
    }

// 依然在 func (pp *partitionProducer) dispatch 中

        } else if pp.highWatermark > 0 {
            // we are retrying something (else highWatermark would be 0) but this message is not a *new* retry level
            if msg.retries < pp.highWatermark {
                // in fact this message is not even the current retry level, so buffer it for now (unless it's a just a fin)
                if msg.flags&fin == fin {
                    pp.retryState[msg.retries].expectChaser = false
                    pp.parent.inFlight.Done() // this fin is now handled and will be garbage collected
                } else {
                // 新消息暂存在buffer中
                    pp.retryState[msg.retries].buf = append(pp.retryState[msg.retries].buf, msg)
                }
                continue


        // 重试的消息会发往新开启的broker中

// 此处逻辑位于 func (bp *brokerProducer) run 方法
            if reason := bp.needsRetry(msg); reason != nil {
                // ”重发”
                bp.parent.retryMessage(msg, reason)
                // clear flag
                if bp.closing == nil && msg.flags&fin == fin {
                    // we were retrying this partition but we can start processing again
                    delete(bp.currentRetries[msg.Topic], msg.Partition)
                    Logger.Printf("producer/broker/%d state change to [closed] on %s/%d\n",
                        bp.broker.ID(), msg.Topic, msg.Partition)
                }

                continue
            }

// 此处逻辑在 func (pp *partitionProducer) dispatch 中
            } else if msg.flags&fin == fin {
                // this message is of the current retry level (msg.retries == highWatermark) and the fin flag is set,
                // meaning this retry level is done and we can go down (at least) one level and flush that
                pp.retryState[pp.highWatermark].expectChaser = false
                pp.flushRetryBuffers()
                pp.parent.inFlight.Done() // this fin is now handled and will be garbage collected
                continue
            }

// 是怎么刷缓存的消息的            
func (pp *partitionProducer) flushRetryBuffers() {
    Logger.Printf("producer/leader/%s/%d state change to [flushing-%d]\n", pp.topic, pp.partition, pp.highWatermark)
    for {
        pp.highWatermark--

        if pp.brokerProducer == nil {
            if err := pp.updateLeader(); err != nil {
                pp.parent.returnErrors(pp.retryState[pp.highWatermark].buf, err)
                goto flushDone
            }
            Logger.Printf("producer/leader/%s/%d selected broker %d\n", pp.topic, pp.partition, pp.leader.ID())
        }
        // 可以看到是一层一层开始刷的
        // 从最外层（即重试次数最多的消息）开始刷起
        for _, msg := range pp.retryState[pp.highWatermark].buf {
            pp.brokerProducer.input <- msg
        }

    flushDone:
        pp.retryState[pp.highWatermark].buf = nil
        if pp.retryState[pp.highWatermark].expectChaser {
            Logger.Printf("producer/leader/%s/%d state change to [retrying-%d]\n", pp.topic, pp.partition, pp.highWatermark)
            break
        } else if pp.highWatermark == 0 {
            Logger.Printf("producer/leader/%s/%d state change to [normal]\n", pp.topic, pp.partition)
            break
        }
    }
}

// client.go的tryRefreshMetadata方法
// 作用是刷新 Metadata
func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int, deadline time.Time) error {
    pastDeadline := func(backoff time.Duration) bool {
        if !deadline.IsZero() && time.Now().Add(backoff).After(deadline) {
            // we are past the deadline
            return true
        }
        return false
    }
    retry := func(err error) error {
        if attemptsRemaining > 0 {
            backoff := client.computeBackoff(attemptsRemaining)
            if pastDeadline(backoff) {
                Logger.Println("client/metadata skipping last retries as we would go past the metadata timeout")
                return err
            }
            Logger.Printf("client/metadata retrying after %dms... (%d attempts remaining)\n", client.conf.Metadata.Retry.Backoff/time.Millisecond, attemptsRemaining)
            if backoff > 0 {
                time.Sleep(backoff)
            }
            return client.tryRefreshMetadata(topics, attemptsRemaining-1, deadline)
        }
        return err
    }

    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
            // valid response, use it
            shouldRetry, err := client.updateMetadata(response, allKnownMetaData)
            if shouldRetry {
                Logger.Println("client/metadata found some partitions to be leaderless")
                return retry(err) // note: err can be nil
            }
            return err
        ....

        case KError:
            // if SASL auth error return as this _should_ be a non retryable err for all brokers
            if err.(KError) == ErrSASLAuthenticationFailed {
                Logger.Println("client/metadata failed SASL authentication")
                return err
            }

            if err.(KError) == ErrTopicAuthorizationFailed {
                Logger.Println("client is not authorized to access this topic. The topics were: ", topics)
                return err
            }
            // else remove that broker and try again
            Logger.Printf("client/metadata got error from broker %d while fetching metadata: %v\n", broker.ID(), err)
            _ = broker.Close()
            client.deregisterBroker(broker)

        default:
            // some other error, remove that broker and try again
            Logger.Printf("client/metadata got error from broker %d while fetching metadata: %v\n", broker.ID(), err)
            _ = broker.Close()
            client.deregisterBroker(broker)
        }
    }

    if broker != nil {
        Logger.Printf("client/metadata not fetching metadata from broker %s as we would go past the metadata timeout\n", broker.addr)
        return retry(ErrOutOfBrokers)
    }

    Logger.Println("client/metadata no available broker to send metadata request to")
    client.resurrectDeadBrokers()
    return retry(ErrOutOfBrokers)
}