case ApiKeys.PRODUCE => handleProduceRequest(request)

def handleProduceRequest(request: RequestChannel.Request) {
  // 将 ByteBuffer 类型的请求解析成 ProduceRequest 类型
  val produceRequest = request.body[ProduceRequest]
  // 取得请求的总字节数, 包含 header 和 body
  val numBytesAppended = request.header.toStruct.sizeOf + request.sizeOfBodyInBytes
  // ...
}

class Request(/* ... */
              val context: RequestContext,
              /* ... */
              @volatile private var buffer: ByteBuffer,
              /* ... */) extends BaseRequest {
  private val bodyAndSize: RequestAndSize = context.parseRequest(buffer)

  def header: RequestHeader = context.header
  def sizeOfBodyInBytes: Int = bodyAndSize.size
  // ...
}

val internalTopicsAllowed = request.header.clientId == AdminUtils.AdminClientId

// call the replica manager to append messages to the replicas
// 传入replicaManager来添加消息到副本上
replicaManager.appendRecords(
  timeout = produceRequest.timeout.toLong, // Produce请求的timeout字段
  requiredAcks = produceRequest.acks, // Produce请求的acks字段
  internalTopicsAllowed = internalTopicsAllowed, // client id是否为__admin_client
  isFromClient = true, // 这里是处理客户端的Produce请求，所以为true
  entriesPerPartition = authorizedRequestInfo, // 通过认证的请求信息
  responseCallback = sendResponseCallback, // 发送响应的回调函数
  processingStatsCallback = processingStatsCallback) // 处理stats的回调函数

// if the request is put into the purgatory, it will have a held reference and hence cannot be garbage collected;
// hence we clear its data here in order to let GC reclaim its memory since it is already appended to log
produceRequest.clearPartitionRecords() // 简单将Produce请求的partitionRecords置为null

val authorizedRequestInfo = mutable.Map[TopicPartition, MemoryRecords]()

// 从 produceRequest.partitionRecords 取得所有 TopicPartiton 和 MemoryRecords
// ***OrFail 方法仅仅检查 partitionRecords 字段是否为 null, 若为 null 则抛出异常
for ((topicPartition, memoryRecords) <- produceRequest.partitionRecordsOrFail.asScala) {
  if (!authorize(request.session, Write, new Resource(Topic, topicPartition.topic)))
    unauthorizedTopicResponses += topicPartition -> new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)
  else if (!metadataCache.contains(topicPartition.topic))
    nonExistingTopicResponses += topicPartition -> new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)
  else // 通过了 authorize 方法认证, 并且 metadataCache 包含该 topic
    authorizedRequestInfo += (topicPartition -> memoryRecords)
}

def appendRecords(timeout: Long,
                  requiredAcks: Short,
                  internalTopicsAllowed: Boolean,
                  isFromClient: Boolean,
                  entriesPerPartition: Map[TopicPartition, MemoryRecords],
                  responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
                  delayedProduceLock: Option[Lock] = None,
                  processingStatsCallback: Map[TopicPartition, RecordsProcessingStats] => Unit = _ => ()) {
  if (isValidRequiredAcks(requiredAcks)) { // acks只能为-1，0，1
    // ...
  } else {
    // acks 在可接受的范围外, 则客户端肯定出错了, 仅仅返回错误, 而不用处理请求
    // 具体处理: 对每个 TopicPartition 对象, 构造相应的 PartitionResponse 对象组成新的 Map
    //  其中包含 error, baseOffset, logAppendTime, logStartOffset 等字段,
    //  除了 error 字段标明为 acks不合法 外, 其余字段都随意设置
    val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
      topicPartition -> new PartitionResponse(Errors.INVALID_REQUIRED_ACKS,
        LogAppendInfo.UnknownLogAppendInfo.firstOffset, RecordBatch.NO_TIMESTAMP, LogAppendInfo.UnknownLogAppendInfo.logStartOffset)
    }
    // 调用传入的回调 responseCallback 将返回值发送回去
    responseCallback(responseStatus)
  }

val sTime = time.milliseconds

class ReplicaManager(val config: KafkaConfig,
                     metrics: Metrics,
                     time: Time,

class KafkaApis(val requestChannel: RequestChannel,
                val replicaManager: ReplicaManager,

apis = new KafkaApis(socketServer.requestChannel, replicaManager, /* ... */)

replicaManager = createReplicaManager(isShuttingDown)

protected def createReplicaManager(isShuttingDown: AtomicBoolean): ReplicaManager =
  new ReplicaManager(config, metrics, time, /* ... */)

class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM,

val sTime = time.milliseconds // 取得当前毫秒级时间戳
val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
  isFromClient = isFromClient, entriesPerPartition, requiredAcks)
// 调试信息: 再次取得时间戳, 相减得到 appendToLocalLog 的用时
debug("Produce to local log in %d ms".format(time.milliseconds - sTime))

private def appendToLocalLog(internalTopicsAllowed: Boolean,
                             isFromClient: Boolean,
                             entriesPerPartition: Map[TopicPartition, MemoryRecords],
                             requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {
  trace(s"Append [$entriesPerPartition] to local log")
  // 遍历所有客户端请求写入的 topicPartition 以及对应消息 records
  entriesPerPartition.map { case (topicPartition, records) =>
    // 更新topicStats，暂时略去不看
    brokerTopicStats.topicStats(topicPartition.topic).totalProduceRequestRate.mark()
    brokerTopicStats.allTopicsStats.totalProduceRequestRate.mark()
    if (Topic.isInternal(topicPartition.topic) && !internalTopicsAllowed) {
      // topic是内部主题: __consumer_offsets 或 __transaction_state, 且 internalTopicsAllowed 为 false
      //  (在 KafkaApis.handleProduceRequest 中, 只有请求的 clientId 为 AdminClientId 时才为 true)
      // 也就是如果不是 Admin 客户端, 尝试写入内部主题则会返回 写入不合法 的 LogAppendResult
      (topicPartition, LogAppendResult(
        LogAppendInfo.UnknownLogAppendInfo,
        Some(new InvalidTopicException(s"Cannot append to internal topic ${topicPartition.topic}"))))
    } else { // 非内部主题, 可以写入
      try {
        // ...
      } catch {
        // 异常处理(略)，会将处理客户端请求的异常信息写入返回结果中
        // 注意，对于用于流程控制的Throwable异常，会单独处理，这里后面再看
      }
    }
  }

// 从内部的 allPartitions 中找到 topicPartition, PS: allPartitions 是从本地消息日志中读取的
val partitionOpt = getPartition(topicPartition)
val info = partitionOpt match {
  case Some(partition) =>
    // 找到的是 OfflinePartition (当前broker不在分区的ISR列表上) 则会(通过异常处理)返回错误信息
    // https://issues.apache.org/jira/browse/KAFKA-6796 在 Kafka 2.0 中对这种行为进行了修复
    // 比如在分区重分配期间, 客户的Produce请求在本地副本被删除后到达, 此时不应该返回分区不存在的错误
    // 因此2.0中抛出的是 NotLeaderForPartitionException, 会强制让客户端更新元数据来找到新的分区位置
    if (partition eq ReplicaManager.OfflinePartition)
      throw new KafkaStorageException(s"Partition $topicPartition is in an offline log directory on broker $localBrokerId")
    // 添加记录到leader副本上
    partition.appendRecordsToLeader(records, isFromClient, requiredAcks)

  // 若找不到目标 topicPartition, 则代表生产者向一个未知的分区生产消息, 返回表示分区不存在的结果
  case None => throw new UnknownTopicOrPartitionException("Partition %s doesn't exist on %d"
    .format(topicPartition, localBrokerId))
}

// 略去更新brokerTopicStats的代码

(topicPartition, LogAppendResult(info))

class Partition(val topic: String,
                val partitionId: Int,
                time: Time,
                replicaManager: ReplicaManager,
                val isOffline: Boolean = false)

def appendRecordsToLeader(records: MemoryRecords, isFromClient: Boolean, requiredAcks: Int = 0): LogAppendInfo = {
  // 用读锁保护, 可以多线程添加记录到 leader副本 上, 但是如果ISR更新过程会获取写锁, 此时要等待ISR更新完毕
  val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
    // 若leader副本的id为本地的broker id, 则返回对应的 Replica对象
    leaderReplicaIfLocal match {
      case Some(leaderReplica) => // leader副本
        val log = leaderReplica.log.get // append-only的Log对象
        val minIsr = log.config.minInSyncReplicas // min.insync.replicas 配置
        val inSyncSize = inSyncReplicas.size // ISR数量

        // acks为-1时, 客户端会等待所有ISR确认收到消息时才返回, 此时配置 min.insync.replicas
        // 指定了这个ISR的最小数量, 因此ISR数量不够时会抛出ISR副本太少的异常
        if (inSyncSize < minIsr && requiredAcks == -1) {
          throw new NotEnoughReplicasException("Number of insync replicas for partition %s is [%d], below required minimum [%d]"
            .format(topicPartition, inSyncSize, minIsr))
        }

        // 将消息集写入日志, 分配 offsets 和 分区leader的epoch
        val info = log.appendAsLeader(records, leaderEpoch = this.leaderEpoch, isFromClient)
        // probably unblock some follower fetch requests since log end offset has been updated
        // 因为 LEO(log end offset) 已经更新了, 所以某些 follower 的 fetch请求可能解除阻塞了, 于是
        // replicaManager.delayedFetchPurgatory 尝试完成该分区的延迟的fetch请求, 因为 LEO(log end offset)已经跟新
        replicaManager.tryCompleteDelayedFetch(TopicPartitionOperationKey(this.topic, this.partitionId))
        // 因为 ISR 可能只剩1个, 因此可能需要增加HW (high watermark)
        (info, maybeIncrementLeaderHW(leaderReplica))

      case None => // 非leader副本
        throw new NotLeaderForPartitionException("Leader not local for partition %s on broker %d"
          .format(topicPartition, localBrokerId))
    }
  }

  // some delayed operations may be unblocked after HW changed
  // 一些延迟操作可能因为 HW 的改变而解除阻塞, 因此尝试完成这些延迟请求
  if (leaderHWIncremented)
    tryCompleteDelayedRequests()

  info // log.AppendAsLeader返回的结果
}

val delayedProducePurgatory: DelayedOperationPurgatory[DelayedProduce],
val delayedFetchPurgatory: DelayedOperationPurgatory[DelayedFetch],
val delayedDeleteRecordsPurgatory: DelayedOperationPurgatory[DelayedDeleteRecords]

DelayedOperationPurgatory[DelayedProduce](
  purgatoryName = "Produce", brokerId = config.brokerId,
  purgeInterval = config.producerPurgatoryPurgeIntervalRequests),
DelayedOperationPurgatory[DelayedFetch](
  purgatoryName = "Fetch", brokerId = config.brokerId,
  purgeInterval = config.fetchPurgatoryPurgeIntervalRequests),
DelayedOperationPurgatory[DelayedDeleteRecords](
  purgatoryName = "DeleteRecords", brokerId = config.brokerId,
  purgeInterval = config.deleteRecordsPurgatoryPurgeIntervalRequests),

class Message(val buffer: ByteBuffer,
              private val wrapperMessageTimestamp: Option[Long] = None,
              private val wrapperMessageTimestampType: Option[TimestampType] = None)

object Message {
  val CrcOffset = 0
  val CrcLength = 4
  val MagicOffset = CrcOffset + CrcLength
  val MagicLength = 1
  val AttributesOffset = MagicOffset + MagicLength
  // ...
}

def checksum: Long = ByteUtils.readUnsignedInt(buffer, CrcOffset)
def magic: Byte = buffer.get(MagicOffset)

def payload: ByteBuffer = sliceDelimited(payloadSizeOffset)
def key: ByteBuffer = sliceDelimited(keySizeOffset)

// 由于key字段长度是动态的，所以无法在object中定义payloadSizeOffset常量，而需要在类中取得key的长度后计算而出
private def payloadSizeOffset = {
  if (magic == MagicValue_V0) KeyOffset_V0 + max(0, keySize)
  else KeyOffset_V1 + max(0, keySize)
}

private def sliceDelimited(start: Int): ByteBuffer = {
  val size = buffer.getInt(start) // 取得前4个字节表示的长度
  if(size < 0) {
    null
  } else {
    // 拷贝一份，防止影响buffer的position，注意这里的拷贝并没拷贝内部的字节序列
    var b = buffer.duplicate()
    // 跳过表示长度的4个字节，回顾协议，key和payload之前都有4个字节表示长度
    b.position(start + 4)
    // 取得buffer从position开始的子序列，并重置长度和position
    // 之后b就是指向key或payload字段并且长度合适的ByteBuffer了
    b = b.slice()
    b.limit(size)
    b.rewind
    b
  }
}

def timestamp: Long = {
  if (magic == MagicValue_V0) // v0版本的消息不使用时间戳
    Message.NoTimestamp
  // 对v1版本的消息，有以下3种Case：
  // 1. 外部消息的时间戳及其类型都为None；
  // 2. 外部消息的时间戳类型为LogAppendTime且时间戳非None；
  // 3. 外部消息的时间戳类型为CreateTime且时间戳非None。
  // Case 2
  else if (wrapperMessageTimestampType.exists(_ == TimestampType.LOG_APPEND_TIME) && wrapperMessageTimestamp.isDefined)
    wrapperMessageTimestamp.get
  else // Case 1, 3
    buffer.getLong(Message.TimestampOffset)
}

def this(bytes: Array[Byte], 
         key: Array[Byte],
         timestamp: Long,
         timestampType: TimestampType,
         codec: CompressionCodec, 
         payloadOffset: Int, 
         payloadSize: Int,
         magicValue: Byte) = {
  // 计算总长度，分配对应大小的ByteBuffer
  this(ByteBuffer.allocate(Message.CrcLength +
                           Message.MagicLength +
                           Message.AttributesLength +
                           (if (magicValue == Message.MagicValue_V0) 0
                            else Message.TimestampLength) +
                           Message.KeySizeLength + 
                           (if(key == null) 0 else key.length) + 
                           Message.ValueSizeLength + 
                           (if(bytes == null) 0 
                            else if(payloadSize >= 0) payloadSize 
                            else bytes.length - payloadOffset)))
  // 验证magic和timestamp是否对应：
  // 1. magic只能为0或1；
  // 2. 时间戳必须为非负数或-1（代表不使用时间戳）；
  // 3. magic为0时timestamp必须为-1，因为v0版本不支持时间戳；
  validateTimestampAndMagicValue(timestamp, magicValue)
  // skip crc, we will fill that in at the end
  // 跳过CRC字段，先填充后面的部分
  buffer.position(MagicOffset)
  buffer.put(magicValue) // 填充 magic
  // 根据压缩类型和时间戳类型计算 attribute 并填充
  val attributes: Byte = LegacyRecord.computeAttributes(magicValue, CompressionType.forId(codec.codec), timestampType)
  buffer.put(attributes)
  // Only put timestamp when "magic" value is greater than 0
  if (magic > MagicValue_V0) // 仅当magic大于0才可以填充时间戳
    buffer.putLong(timestamp)
  if(key == null) { // key为空则将keylen填充为-1，代表key不存在
    buffer.putInt(-1)
  } else { // 否则填充 keylen 和 key
    buffer.putInt(key.length)
    buffer.put(key, 0, key.length)
  }
  // 类似key，若bytes为空，填充len为-1，否则填充 bytes 的指定部分
  // payloadOffset指定起始偏移量，payloadSize指定填充字节数(若<0则填充偏移量之后的所有字节)
  val size = if(bytes == null) -1
             else if(payloadSize >= 0) payloadSize 
             else bytes.length - payloadOffset
  buffer.putInt(size)
  if(bytes != null)
    buffer.put(bytes, payloadOffset, size)
  buffer.rewind() // 重置position为初始，以便之后getXXX()读取

  // now compute the checksum and fill it in
  // 后面字段填充完了，计算CRC校验值并填充到前4个字节，完成后position为4，
  // 也就是对内部buffer调用slice()方法返回的是magic字段至末尾的部分而不包含CRC字段
  ByteUtils.writeUnsignedInt(buffer, CrcOffset, computeChecksum)
}

// object Message
def fromRecord(record: LegacyRecord): Message = {
  val wrapperTimestamp: Option[Long] = if (record.wrapperRecordTimestamp == null) None else Some(record.wrapperRecordTimestamp)
  val wrapperTimestampType = Option(record.wrapperRecordTimestampType)
  new Message(record.buffer, wrapperTimestamp, wrapperTimestampType)
}

// class Message
private[message] def asRecord: LegacyRecord = wrapperMessageTimestamp match {
  case None => new LegacyRecord(buffer)
  case Some(timestamp) => new LegacyRecord(buffer, timestamp, wrapperMessageTimestampType.orNull)
}

private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)

// 取得下一个请求，或者阻塞直到超时  
def receiveRequest(timeout: Long): RequestChannel.BaseRequest =
  requestQueue.poll(timeout, TimeUnit.MILLISECONDS)

val req = requestChannel.receiveRequest(300) // 300ms超时

val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)

var requestHandlerPool: KafkaRequestHandlerPool = null

var socketServer: SocketServer = null

requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId,
                                                 socketServer.requestChannel,
                                                 apis, time, config.numIoThreads)

for (i <- 0 until numThreads) {
  createHandler(i)
}

def createHandler(id: Int): Unit = synchronized {
  runnables += new KafkaRequestHandler(id, brokerId, aggregateIdleMeter, threadPoolSize, requestChannel, apis, time)
  KafkaThread.daemon("kafka-request-handler-" + id, runnables(id)).start()
}

def run() {
  while (!stopped) {
    // 从requestChannel中取得请求，timeout为300ms
    val req = requestChannel.receiveRequest(300)

    req match {
      // Shutdown类型的请求，直接退出Handler线程
      case RequestChannel.ShutdownRequest =>
        debug(s"Kafka request handler $id on broker $brokerId received shut down command")
        shutdownComplete.countDown()
        return

      // 正常请求
      case request: RequestChannel.Request =>
        try {
          request.requestDequeueTimeNanos = endTime
          // 交给apis处理
          apis.handle(request)
        } catch {  // 处理api.handle()可能抛出的异常
          case e: FatalExitError =>
            shutdownComplete.countDown()
            Exit.exit(e.statusCode)
          case e: Throwable => error("Exception when handling request", e)
        } finally {
          request.releaseBuffer()
        }

      case null => // continue
    }
  }
  shutdownComplete.countDown()
}

def sendShutdownRequest(): Unit = requestQueue.put(ShutdownRequest)

def initiateShutdown(): Unit = requestChannel.sendShutdownRequest()

def shutdown(): Unit = synchronized {
  info("shutting down")
  for (handler <- runnables)
    handler.initiateShutdown()
  for (handler <- runnables)
    handler.awaitShutdown()
}

private val shutdownComplete = new CountDownLatch(1)

def handle(request: RequestChannel.Request) {
  try {
    request.header.apiKey match {
      case ApiKeys.PRODUCE => handleProduceRequest(request)
      case ApiKeys.FETCH => handleFetchRequest(request)
      // 其他类型的的ApiKeys...
    }
  } catch {
    case e: FatalExitError => throw e
    case e: Throwable => handleError(request, e)
  } finally {
    request.apiLocalCompleteTimeNanos = time.nanoseconds
  }
}

def handleListOffsetRequest(request: RequestChannel.Request) {
  val version = request.header.apiVersion()

  // 根据请求头的API版本进行不同的处理
  val mergedResponseMap = if (version == 0)
    handleListOffsetRequestV0(request)
  else
    handleListOffsetRequestV1AndAbove(request)

  // 参数2为将requestThrottleMs作为输入的函数，会根据该输入创建ListOffsetResponse对象
  // 并且在sendResponseMaybeThrottle类根据requestThrottleMs判断是否创建该对象并发送
  sendResponseMaybeThrottle(request, requestThrottleMs => new ListOffsetResponse(requestThrottleMs, mergedResponseMap.asJava))
}

private def sendResponse(request: RequestChannel.Request, responseOpt: Option[AbstractResponse]): Unit = {
  // 对响应的每个非0错误码更新度量指标
  responseOpt.foreach(response => requestChannel.updateErrorMetrics(request.header.apiKey, response.errorCounts.asScala))

  responseOpt match {
    case Some(response) =>
      // 若响应存在，则构造响应的Send，并调用requestChannel的sendResponse()方法发送
      val responseSend = request.context.buildResponse(response)
      // 如果RequestChannel伴生对象的isRequestloggingEnabled则构造该字符串
      val responseString =
        if (RequestChannel.isRequestLoggingEnabled) Some(response.toString(request.context.apiVersion))
        else None
      requestChannel.sendResponse(new RequestChannel.Response(request, Some(responseSend), SendAction, responseString))
    case None =>
      requestChannel.sendResponse(new RequestChannel.Response(request, None, NoOpAction, None))
  }
}

object RequestChannel extends Logging {
  private val requestLogger = Logger("kafka.request.logger")
  // ...

  def isRequestLoggingEnabled: Boolean = requestLogger.underlying.isDebugEnabled
  // ...
}

val header = RequestHeader.parse(receive.payload)
val context = new RequestContext(header, receive.source, channel.socketAddress,
  channel.principal, listenerName, securityProtocol)
val req = new RequestChannel.Request(processor = id, context = context,
  startTimeNanos = time.nanoseconds, memoryPool, receive.payload, requestChannel.metrics)
requestChannel.sendRequest(req)

// 发送待处理的请求, requestQueue 有容量上限, 由 queue.max.requests 配置, 若达到了上限则会阻塞。
def sendRequest(request: RequestChannel.Request) {
  requestQueue.put(request)
}

val context = new RequestContext(header, receive.source, channel.socketAddress,  channel.principal,
                                 listenerName, securityProtocol)

def body[T <: AbstractRequest](implicit classTag: ClassTag[T], nn: NotNothing[T]): T = {
  bodyAndSize.request match {
    case r: T => r
    case r =>
      throw new ClassCastException(s"Expected request with type ${classTag.runtimeClass}, but found ${r.getClass}")
  }
}

private val bodyAndSize: RequestAndSize = context.parseRequest(buffer)

// 将请求的 ByteBuffer 解析成 MetadataRequest 对象
val metadataRequest = request.body[MetadataRequest]

// 取得下个请求, 或者阻塞直到超时
def receiveRequest(timeout: Long): RequestChannel.BaseRequest =
  requestQueue.poll(timeout, TimeUnit.MILLISECONDS)

// 类 KafkaRequestHandler 的方法
def run() {
  while (!stopped) {
    // ...
    val req = requestChannel.receiveRequest(300) // 300ms超时
    // ...
    req match {
      case RequestChannel.ShutdownRequest =>
        // ...
      case request: RequestChannel.Request =>
        // ...
      case null => // continue
    }
  }
  // ...
}

def sendResponse(response: RequestChannel.Response) {
  if (isTraceEnabled) {
    // 判断响应类型并打印日志(略)
  }

  val processor = processors.get(response.processor)
  // 如果 processor 已经关闭了, 可能会被移出 processors (此时返回null), 因此直接丢掉响应
  if (processor != null) {
    processor.enqueueResponse(response)
  }
}

private def sendResponse(request: RequestChannel.Request, responseOpt: Option[AbstractResponse]): Unit = {
  // 更新metrics(略)

  responseOpt match {
    case Some(response) =>
      val responseSend = request.context.buildResponse(response)
      val responseString =
        if (RequestChannel.isRequestLoggingEnabled) Some(response.toString(request.context.apiVersion))
        else None
      requestChannel.sendResponse(new RequestChannel.Response(request, Some(responseSend), SendAction, responseString))
    case None =>
      requestChannel.sendResponse(new RequestChannel.Response(request, None, NoOpAction, None))
  }
}

private val startupLatch = new CountDownLatch(1)
@volatile private var shutdownLatch = new CountDownLatch(0)

private val alive = new AtomicBoolean(true)
protected def isRunning: Boolean = alive.get

def shutdown(): Unit = {
  // 如果线程仍在运行, 则将 alive 置为true表示线程之后会关闭, 然后调用抽象方法 wakeup()
  if (alive.getAndSet(false))
    wakeup()
  shutdownLatch.await()
}

// 等待线程完全启动
def awaitStartup(): Unit = startupLatch.await

// 标识线程已经启动, 这样就可以等待停止操作了, 因此将 shutdownLatch 指向倒计时为1的对象
// 这样做是为了防止启动时抛出异常, 比如绑定正在使用的地址, 此时应该在处理异常之后仍然能 shutdown,
// 此时 shutdownComplete 的调用会因为异常而被跳过, 如果计数初始化为1会一直阻塞
protected def startupComplete(): Unit = {
  // Replace the open latch with 
  shutdownLatch = new CountDownLatch(1)
  startupLatch.countDown()
}

// 标识线程已经关闭
protected def shutdownComplete(): Unit = shutdownLatch.countDown()

def close(channel: SocketChannel): Unit = {
  if (channel != null) {
    debug("Closing connection from " + channel.socket.getRemoteSocketAddress())
    // 减少 channel 对应地址的连接计数
    connectionQuotas.dec(channel.socket.getInetAddress)
    // 关闭 socket 连接以及 channel 本身, 吞下异常, 也就是说关闭出错不是什么严重错误, 写入日志以供分析就行
    CoreUtils.swallow(channel.socket().close(), this, Level.ERROR)
    CoreUtils.swallow(channel.close(), this, Level.ERROR)
  }
}

def run() {
  serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT) // 注册OP_ACCEPT事件
  // 标识启动完成, 之后 acceptor.awaitStartup() 才会返回, 回顾 createAcceptorAndProcessors()
  // 也就是说 Server 启动时, 必须等到 acceptor 注册 OP_ACCEPT 事件后才会执行后续步骤:
  //   将acceptor加入acceptors, addProcessors, 创建下个acceptor ...
  startupComplete()
  try {
    var currentProcessor = 0 // 记录当前processor的id
    while (isRunning) {
      try {
        // 轮询Selector直到有channels准备好I/O, 或者超时(500ms)
        val ready = nioSelector.select(500)
        if (ready > 0) {  // 有ready个channels准备好I/O
          // TODO: 处理准备好I/O的channels
        }  // else: ready <= 0
      }
      catch {
        // 假设有特定的channel在select时出错, 或者收到bad request, 我们不想要让其他channels受到影响
        // 因此遇到异常只需要打印错误即可。
        // 但是scala会通过ControlThrowable来进行流程控制, 所以此时需要继续将异常往上抛(这是安全的)
        // 在scala 2.13中可以用 case NonFatal(e) 来避免ControlThrowable被捕获
        case e: ControlThrowable => throw e
        case e: Throwable => error("Error occurred", e)
      }
    }
  } finally {
    // Acceptor线程结束后的清理工作
    debug("Closing server socket and selector.")
    CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)
    CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)
    shutdownComplete()
  }
}

try {
  // 遍历所有的key, 类型为SelectionKey
  val key = iter.next
  iter.remove() // 从集合中移除该key, 防止
  if (key.isAcceptable) {  // 该key的channel可以接受新的socket连接
    // round robin算法, 将连接均衡分配给计算出的下标对应的processor
    // 比如3个processors, 接收了7个连接, 则分配的processor下标依次为: 0,1,2,0,1,2,0,1
    val processor = synchronized {
      currentProcessor = currentProcessor % processors.size
      processors(currentProcessor)
    }
    accept(key, processor)
  } else
    throw new IllegalStateException("Unrecognized key state for acceptor thread.")

  // round robin算法, 迭代
  currentProcessor = currentProcessor + 1
} catch { // 遍历keys及处理每个key时的异常在此打印
  case e: Throwable => error("Error while accepting connection", e)
}

def accept(key: SelectionKey, processor: Processor) {
  // key.channel()向下转型, 从抽象类 SelectableChannel 转型为派生类 ServerSocketChannel
  val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
  val socketChannel = serverSocketChannel.accept() // socket accept, 返回表示连接的 SocketChannel
  try {
    // 将远程地址对应的连接数加1, 如果超过了配置的最大连接数限额, connectionQuotas会抛出 TooManyConnectionsException
    connectionQuotas.inc(socketChannel.socket().getInetAddress)
    socketChannel.configureBlocking(false) // socket设为非阻塞模式
    socketChannel.socket().setTcpNoDelay(true) // socket设置TCP_NODELAY选项, 禁止Nagle算法
    socketChannel.socket().setKeepAlive(true) // socket设置保活模式, 长时间没有发送心跳则发出RST包重置连接
    // 配置的 socket.send.buffer.bytes 不为默认值, 则设置 SO_SNDBUF 选项重置发送缓冲区大小
    if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
      socketChannel.socket().setSendBufferSize(sendBufferSize)

    processor.accept(socketChannel)
  } catch {
    case e: TooManyConnectionsException =>
      close(socketChannel)
  }
}

def accept(socketChannel: SocketChannel) {
  newConnections.add(socketChannel)
  wakeup()
}

override def run() {
  startupComplete()
  try {
    while (isRunning) {
      try {
        configureNewConnections() // 处理缓存的新连接
        processNewResponses() // 处理缓存的响应
        poll() // 轮询, 从 Selector 中获取准备好I/O的事件
        // 处理已完成的接收/发送以及断开的channels
        processCompletedReceives()
        processCompletedSends()
        processDisconnected()
      } catch {
        // 这里吞下了所有异常, 因为让 processor 退出对 broker 的影响可能会很大, 但值得商榷的是,
        // 是否存在需要让整个 broker 停止的异常。
        // 通常抛出的异常都是和特定socket或者bad request相关的, 这些异常被捕获, 然后会被独立的方法处理, 因此不会在这里
        // 被捕获, 所以可能这里只会看到 ControlThrowable (仅仅是可见, 没有地方会抛出)
        case e: Throwable => processException("Processor got uncaught exception.", e)
      }
    }
  } finally {
    // 将异常信息写入日志(略)
    shutdownComplete()
  }
}

private def configureNewConnections() {
  while (!newConnections.isEmpty) {
    // newConnections的SocketChannel依次出队
    val channel = newConnections.poll()
    try {
      // 调用链: selector.registerChannel => selector.buildAndAttachKafkaChannel => channelBuilder.buildChannel()
      // channel 会注册 OP_READ 事件(返回 SelectionKey)到 selector 上, 然后和 connectionId, SelectionKey 一起构造
      // KafkaChannel 对象, 以 connectionId 作为key组成键值对加入 selector.channels 中
      selector.register(connectionId(channel.socket), channel)
    } catch {
      // 捕获所有异常, 关闭 对应的socket防止socket泄漏
      case e: Throwable =>
        close(channel)
        // 将异常信息写入日志(略)
    }
  }
}

// id: SocketChannel.connectionId
// key: SocketChannel.register() 返回的 SelectionKey
// maxReceiveSize: config.socketRequestMaxBytes, 即配置"socket.request.max.bytes"
// memoryPool: SocketServer.memoryPool
KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize, memoryPool);
key.attach(channel);  // key原本是attach之前的SocketChannel的, 现在改变attach的对象

private def processNewResponses() {
  var curr: RequestChannel.Response = null
  while ({curr = dequeueResponse(); curr != null}) {
    // 将响应从 responseQueue 中依次出队, 这里取得 connectionId 作为 channelId
    val channelId = curr.request.context.connectionId
    try {
      // 根据响应的类型进行不同操作
      curr.responseAction match {
        case RequestChannel.NoOpAction =>
          // 无操作: 无需发送响应给客户端, 因此需要读取更多请求到服务端的socket buffer中
          // 调用链: selector.unmute() => channel.unmute()
          // 会将 channel 从 selector.explicitlyMutedChannels 中移除,
          // 如果该channel处于连接状态, 会在 channel.transportLayer 注册 OP_READ 事件。
          updateRequestMetrics(curr)
          trace("Socket server received empty response to send, registering for read: " + curr)
          openOrClosingChannel(channelId).foreach(c => selector.unmute(c.id))
        case RequestChannel.SendAction =>
          // 发送: 调用链为 sendResponse() => selector.send() => channel.setSend()
          // 将响应加入 inflightResponses 中, 并在 channel.transportLayer 注册 OP_WRITE 事件
          val responseSend = curr.responseSend.getOrElse(
            throw new IllegalStateException(s"responseSend must be defined for SendAction, response: $curr"))
          sendResponse(curr, responseSend)
        case RequestChannel.CloseConnectionAction =>
          // 关闭连接： 关闭channel
          updateRequestMetrics(curr)
          trace("Closing socket connection actively according to the response code.")
          close(channelId)
      }
    } catch {
      // 将异常信息写入日志(略)
    }
  }
}

private def poll() {
  // 轮询300ms, 会将读取的请求/发送的响应/断开的连接，放入 selector 的 completeReceives/completedSends/disconnected
  try selector.poll(300)
  catch {
    case e @ (_: IllegalStateException | _: IOException) =>
      // 不会重抛异常, 这样这次轮询的所有完成的 sends/receives/connections/disconnections 事件都会被处理
      error(s"Processor $id poll failed due to illegal state or IO exception")
  }
}

// 类 KafkaChannel
// 调用链: Selector.PollSelectionKeys() => write() => send()
private boolean send(Send send) throws IOException {
    send.writeTo(transportLayer);
    if (send.completed())
        transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

    return send.completed();
}

private def processCompletedReceives() {
  // 遍历所有完成接收的 NetworkService, 具体实现在 selector.poll() 方法中, 最后会调用 addToCompletedReceives()
  // 如果 channel 不在 explicitlyMutedChannels 中 (即调用了unmute()方法), 则会将 channel 对应的 NetworkService 队列
  // 弹出队首并加入 completedReceives 中。
 selector.completedReceives.asScala.foreach { receive =>
    try {
      // NetworkServer 的 source 字段记录了连接channel的 connectionId
      openOrClosingChannel(receive.source) match {
        case Some(channel) =>
          // 解析 payload (接收到的ByteBuffer)的头部
          val header = RequestHeader.parse(receive.payload)
          // 将其与 channel 的会话层信息封装成 RequestContext
          val context = new RequestContext(header, receive.source, channel.socketAddress,
            channel.principal, listenerName, securityProtocol)
          // 进一步将上述信息封装成 Request 对象
          val req = new RequestChannel.Request(processor = id, context = context,
            startTimeNanos = time.nanoseconds, memoryPool, receive.payload, requestChannel.metrics)
          // 这里仅仅是将 req 放入 requestChannel 的内部队列 requestQueue
          requestChannel.sendRequest(req)
          // 取消监听该channel的 OP_READ 事件, 并添加到 explicitlyMutedChannels
          selector.mute(receive.source)
        case None =>
          // 抛出异常(略)
      }
    } catch {
      // 异常处理(略)
    }
  }
}

private def processCompletedSends() {
  // 遍历所有完成发送的 NetworkService, 具体实现在 selector.poll() 方法中
  selector.completedSends.asScala.foreach { send =>
    try {
      // 将该网络地址的响应从 inflightResponses 中移除
      val resp = inflightResponses.remove(send.destination).getOrElse {
        throw new IllegalStateException(s"Send for ${send.destination} completed, but not in `inflightResponses`")
      }
      updateRequestMetrics(resp)
      // 将对应的 channel 从 explicitlyMutedChannels 中移除, 并且如果未断开连接, 则注册 OP_READ 事件
      selector.unmute(send.destination)
    } catch {
      // 异常处理(略)
    }
  }

private def processDisconnected() {
  // 遍历所有断开连接的channel的 connectionId, 具体实现在 selector.poll() 方法中
  selector.disconnected.keySet.asScala.foreach { connectionId =>
    try {
      // 从 connectionId 中取得网络地址信息
      val remoteHost = ConnectionId.fromString(connectionId).getOrElse {
        throw new IllegalStateException(s"connectionId has unexpected format: $connectionId")
      }.remoteHost
      // 将断开连接的网络地址的响应从 inflightResponses 中移除
      inflightResponses.remove(connectionId).foreach(updateRequestMetrics)
      // the channel has been closed by the selector but the quotas still need to be updated
      // 更新 quotas 的信息, 即将该网络地址上的连接数减1
      connectionQuotas.dec(InetAddress.getByName(remoteHost))
    } catch {
      // 异常处理(略)
    }
  }
}

val config: KafkaConfig // 配置文件
val metric: Metrics // 度量指标
val time: Time // 对象创建时间
val credentialProvider: CredentialProvider // 证书提供者

private val logContext = new LogContext(s"[SocketServer brokerId=${config.brokerId}] ")
this.logIdent = logContext.logPrefix

// 每个ip的最大连接数, 配置: max.connections.per.ip
private val maxConnectionsPerIp = config.maxConnectionsPerIp
// 指定ip的最大连接数, 会覆盖 maxConnectionsPerIp, 配置: max.connections.per.ip.overrides
private val maxConnectionsPerIpOverrides = config.maxConnectionsPerIpOverrides
// 由以上参数创建
private var connectionQuotas: ConnectionQuotas = _

// 请求队列的最大容量, 配置: queued.max.requests
private val maxQueuedRequests = config.queuedMaxRequests
// 内部维护了一个请求队列
val requestChannel = new RequestChannel(maxQueuedRequests)

// 每个 Processor 拥有自己的 Selector, 用于从连接中读取请求和写回响应
// Processor 会将请求发送至 requestChannel, 会从 responseChannels 中读取响应
private val processors = new ConcurrentHashMap[Int, Processor]() // key: id
private var nextProcessorId = 0 // 递增作为每个 processor 的id

// key: EndPoint, 即配置 listeners 指定的 ip/port 以及其 SecurityProtocol (默认PLAINTEXT)
// 对每个绑定的 ip/port 都创建唯一对应的 Acceptor, 用于创建连接 Channel
// PS: 该字段在 network包 内可访问，实际上目前也就这个类会访问。
private[network] val acceptors = new ConcurrentHashMap[EndPoint, Acceptor]()

def startup() {
  this.synchronized {
    // connectionQuotas 用于限制IP的最大连接数
    connectionQuotas = new ConnectionQuotas(maxConnectionsPerIp, maxConnectionsPerIpOverrides)
    // 创建 Acceptors 和 Processors
    createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)
  }

  // 忽略了剩下的代码, 都是调用 KafkaMetricsGroup.newGauge 创建 Gauge对象, 
  // 用于测量某些指标, 比如Processor的平均闲置百分比/内存池可用大小/内存池占用大小
}

private def createAcceptorAndProcessors(processorsPerListener: Int,
                                        endpoints: Seq[EndPoint]): Unit = synchronized {

  // socket内部缓冲区大小，底层可用C函数 setsockopt 设置 SO_SNDBUF/SO_RCVBUF
  val sendBufferSize = config.socketSendBufferBytes // socket.send.buffer.bytes
  val recvBufferSize = config.socketReceiveBufferBytes // socket.receive.buffer.bytes
  val brokerId = config.brokerId // broker.id

  endpoints.foreach { endpoint =>
    val listenerName = endpoint.listenerName
    val securityProtocol = endpoint.securityProtocol

    val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)
    KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()
    acceptor.awaitStartup() // 等待acceptor线程完全启动
    acceptors.put(endpoint, acceptor) // 构建键值对 <EndPoint, Acceptor> 加入 acceptors
    // 单个 Acceptor 对应 processorsPerListener(即numNetworkThreads)个 Processors
    addProcessors(acceptor, endpoint, processorsPerListener)
  }
}

private def addProcessors(acceptor: Acceptor, endpoint: EndPoint, newProcessorsPerListener: Int): Unit = synchronized {
  val listenerName = endpoint.listenerName
  val securityProtocol = endpoint.securityProtocol
  val listenerProcessors = new ArrayBuffer[Processor]()

  for (_ <- 0 until newProcessorsPerListener) {
    val processor = newProcessor(nextProcessorId, connectionQuotas, listenerName, securityProtocol, memoryPool)
    listenerProcessors += processor
    // 内部调用了 putIfAbsent() 方法将其加入了 requestChannel 内部维护的 ConcurrentHashMap[Int, Processor],
    // key为 processor.id, 使用 putIfAbsent() 方法是因为理论上 processer.id 是唯一的, 因此在插入重复的id时,
    // 不应插入新对象, 而是仅仅返回一个非空 Processor 并根据返回值打印日志
    requestChannel.addProcessor(processor)
    // 仅在此递增 SocketServer 的 nextProcessId 字段，因此保证了对不同 Acceptor 的 Processors,
    // 其 id 是不同的，因此每个 Processor 在 processors 字段中都对应唯一的 key
    nextProcessorId += 1
  }
  listenerProcessors.foreach(p => processors.put(p.id, p))
  // 启动 listenerProcessors 的所有 Processor 线程, 并将其添加到内部维护的 processor: ArrayBuffer[Processor] 中
  acceptor.addProcessors(listenerProcessors)
}

// 创建了固定长度的请求队列，queueSize由 queued.max.requests 决定
private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
// 之前已经提过，创建 Processors 时添加进的
private val processors = new ConcurrentHashMap[Int, Processor]()

// NIO Selector, 用于注册 connect, read, write 等事件，并将事件分发给 Acceptor, Processors
private val nioSelector = NSelector.open()
// 服务通道, 绑定 EndPoint, 用于接收客户端的连接, 类型为 ServerSocketChannel
val serverChannel = openServerSocket(endPoint.host, endPoint.port)
// 之前提过的，保存每个 Acceptor 对应的 Processors, 没有存为映射, 因为 Acceptor 要将
// 连接均衡地分配给 Processors, 不涉及查询操作, 更多地需要遍历, 比如round-robin算法
private val processors = new ArrayBuffer[Processor]()

import java.nio.channels.{Selector => NSelector}

// SocketChannel 的并发队列, 用于管理 Acceptor 分配的socket连接
private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()
// ConnectionId 到 Response 的映射, 缓存待发送的响应
private val inflightResponses = mutable.Map[String, RequestChannel.Response]()
// 缓存产生的响应
private val responseQueue = new LinkedBlockingDeque[RequestChannel.Response]()

// 创建的是 kafka.common.network.Selector, 也就是自己实现的 Selector
private val selector = createSelector(
  ChannelBuilders.serverChannelBuilder(listenerName,
    listenerName == config.interBrokerListenerName,
    securityProtocol,
    config,
    credentialProvider.credentialCache,
    credentialProvider.tokenCache))

// 用于生成连接的 index, 类似 SocketServer 生成 Processor.id, 不过保证了 index 非负
private var nextConnectionIndex = 0

// 作为 inflightResponses 的key, 记录了本地地址/远程地址, 以及连接对应的索引
// 该索引是通过 nextConnectionIndex 自增生成的, 而且非负
private[network] case class ConnectionId(localHost: String, localPort: Int, remoteHost: String, remotePort: Int, index: Int) {
  override def toString: String = s"$localHost:$localPort-$remoteHost:$remotePort-$index"
}

private val processors = new Array[Processor](totalProcessorThreads)
private[network] val acceptors = mutable.Map[EndPoint, Acceptor]()