Kafka源码系列之kafka如何实现高性能读写的

def partitionAndCollate(messages: Seq[KeyedMessage[K,Message]]): Option[Map[Int, collection.mutable.Map[TopicAndPartition, Seq[KeyedMessage[K,Message]]]]] = {
 val ret = new HashMap[Int, collection.mutable.Map[TopicAndPartition, Seq[KeyedMessage[K,Message]]]]
 try {
 for (message <- messages) {
 //返回值是指定topic的所有分区在broker分布 (brokerId, numPartitions)
 val topicPartitionsList = getPartitionListForTopic(message)
 //假如指定了分区方法和key，会调用我们的分区方法，获取key指定的分区
 val partitionIndex = getPartition(message.topic, message.partitionKey, topicPartitionsList)
 //获取去指定分区的PartitionAndLeader
 val brokerPartition = topicPartitionsList(partitionIndex)

 // postpone the failure until the send operation, so that requests for other brokers are handled correctly
 val leaderBrokerId = brokerPartition.leaderBrokerIdOpt.getOrElse(-1)

 var dataPerBroker: HashMap[TopicAndPartition, Seq[KeyedMessage[K,Message]]] = null

 //[Int, collection.mutable.Map[TopicAndPartition, Seq[KeyedMessage[K,Message]]]]
      // 获取数组[TopicAndPartition, Seq[KeyedMessage[K,Message]]]]，不存在的话创建
 ret.get(leaderBrokerId) match {
 case Some(element) =>
          dataPerBroker = element.asInstanceOf[HashMap[TopicAndPartition, Seq[KeyedMessage[K,Message]]]]
 case None =>
          dataPerBroker = new HashMap[TopicAndPartition, Seq[KeyedMessage[K,Message]]]
          ret.put(leaderBrokerId, dataPerBroker)
      }

 val topicAndPartition = TopicAndPartition(message.topic, brokerPartition.partitionId)
 var dataPerTopicPartition: ArrayBuffer[KeyedMessage[K,Message]] = null

 //取出[TopicAndPartition, Seq[KeyedMessage[K,Message]]]]中的消息数组Seq[KeyedMessage[K,Message]]]，将消息加入
 dataPerBroker.get(topicAndPartition) match {
 case Some(element) =>
          dataPerTopicPartition = element.asInstanceOf[ArrayBuffer[KeyedMessage[K,Message]]]
 case None =>
          dataPerTopicPartition = new ArrayBuffer[KeyedMessage[K,Message]]
          dataPerBroker.put(topicAndPartition, dataPerTopicPartition)
      }
      dataPerTopicPartition.append(message)
    }
 Some(ret)

val messagesPerTopicPartition = messagesPerTopicAndPartition.map { case (topicAndPartition, messages) =>
 val rawMessages = messages.map(_.message)
  ( topicAndPartition,
 config.compressionCodec match {
 case NoCompressionCodec =>
        debug("Sending %d messages with no compression to %s".format(messages.size, topicAndPartition))
 new ByteBufferMessageSet(NoCompressionCodec, rawMessages: _*)
 case _ =>
        config.compressedTopics.size match {
 case 0 =>
            debug("Sending %d messages with compression codec %d to %s"
 .format(messages.size, config.compressionCodec.codec, topicAndPartition))
 new ByteBufferMessageSet(config.compressionCodec, rawMessages: _*)
 case _ =>
 if(config.compressedTopics.contains(topicAndPartition.topic)) {
              debug("Sending %d messages with compression codec %d to %s"
 .format(messages.size, config.compressionCodec.codec, topicAndPartition))
 new ByteBufferMessageSet(config.compressionCodec, rawMessages: _*)
            }
 else {
              debug("Sending %d messages to %s with no compression as it is not in compressed.topics - %s"
 .format(messages.size, topicAndPartition, config.compressedTopics.toString))
 new ByteBufferMessageSet(NoCompressionCodec, rawMessages: _*)
            }
        }
    }
    )
}
messagesPerTopicPartition

compressionCodec match {
 case DefaultCompressionCodec => new GZIPOutputStream(stream)
 case GZIPCompressionCodec => new GZIPOutputStream(stream)
 case SnappyCompressionCodec => 
 import org.xerial.snappy.SnappyOutputStream
 new SnappyOutputStream(stream)
 case LZ4CompressionCodec =>
 new KafkaLZ4BlockOutputStream(stream)
 case _ =>
 throw new kafka.common.UnknownCodecException("Unknown Codec: " + compressionCodec)
}

case RequestKeys.ProduceKey => handleProducerOrOffsetCommitRequest(request)
val localProduceResults = appendToLocalLog(produceRequest, offsetCommitRequestOpt.nonEmpty)

partitionAndData.map {case (topicAndPartition, messages) =>
 try {
 if (Topic.InternalTopics.contains(topicAndPartition.topic) &&
        !(isOffsetCommit && topicAndPartition.topic == OffsetManager.OffsetsTopicName)) {
 throw new InvalidTopicException("Cannot append to internal topic %s".format(topicAndPartition.topic))
    }
 //主要是获取到分区
 val partitionOpt = replicaManager.getPartition(topicAndPartition.topic, topicAndPartition.partition)
 val info = partitionOpt match {
 case Some(partition) =>
 //开始追加到日志
 partition.appendMessagesToLeader(messages.asInstanceOf[ByteBufferMessageSet],producerRequest.requiredAcks)

val leaderReplicaOpt = leaderReplicaIfLocal()
leaderReplicaOpt match {
 case Some(leaderReplica) =>
 val log = leaderReplica.log.get
 

val appendInfo = analyzeAndValidateMessageSet(messages)

val offset = new AtomicLong(nextOffsetMetadata.messageOffset)
 try {
//            给我们的数据块赋值偏移
 validMessages = validMessages.assignOffsets(offset, appendInfo.codec)

// maybe roll the log if this segment is full
val segment = maybeRoll(validMessages.sizeInBytes)
// now append to the log 并未拆解 初始偏移和消息块
segment.append(appendInfo.firstOffset, validMessages)
// increment the log end offset //更新修大偏移
updateLogEndOffset(appendInfo.lastOffset + 1)

def append(messages: ByteBufferMessageSet) {
 val written = messages.writeTo(channel, 0, messages.sizeInBytes)
 _size.getAndAdd(written)
}

case RequestKeys.FetchKey => handleFetchRequest(request)
val fetchRequest = request.requestObj.asInstanceOf[FetchRequest]
val dataRead = replicaManager.readMessageSets(fetchRequest)
在readMessageSets方法中
//读取指定topic 分区 offset 最大获取大小 MessageSet
val (fetchInfo, highWatermark) = readMessageSet(topic, partition, offset, fetchSize, fetchRequest.replicaId)

val localReplica = if(fromReplicaId == Request.DebuggingConsumerId)
 getReplicaOrException(topic, partition)
else
 getLeaderReplicaIfLocal(topic, partition)
trace("Fetching log segment for topic, partition, offset, size = " + (topic, partition, offset, maxSize))
val maxOffsetOpt =
 if (Request.isValidBrokerId(fromReplicaId))
    None
 else
 Some(localReplica.highWatermark.messageOffset) //最大消息偏移
val fetchInfo = localReplica.log match {
 case Some(log) =>
    log.read(offset, maxSize, maxOffsetOpt)

var entry = segments.floorEntry(startOffset)
val fetchInfo = entry.getValue.read(startOffset, maxOffset, maxLength)

val logSize = log.sizeInBytes // this may change, need to save a consistent copy
val startPosition = translateOffset(startOffset) //根据offset找到物理存储的偏移的位置

// if the start position is already off the end of the log, return null
if(startPosition == null)
 return null

val offsetMetadata = new LogOffsetMetadata(startOffset, this.baseOffset, startPosition.position)

// if the size is zero, still return a log segment but with zero size
if(maxSize == 0)
 return FetchDataInfo(offsetMetadata, MessageSet.Empty)

// calculate the length of the message set to read based on whether or not they gave us a maxOffset
val length = 
  maxOffset match {
 case None =>
 // no max offset, just use the max size they gave unmolested 无麻烦的
 maxSize
 case Some(offset) => {
 // there is a max offset, translate it to a file position and use that to calculate the max read size
 if(offset < startOffset)
 throw new IllegalArgumentException("Attempt to read with a maximum offset (%d) less than the start offset (%d).".format(offset, startOffset))

 val mapping = translateOffset(offset, startPosition.position)
 val endPosition = 
 if(mapping == null) //说明消息最大偏移超过了当前块，所以最大偏移就是当前块大小
 logSize // the max offset is off the end of the log, use the end of the file
 else
 mapping.position //否则就是最大偏移
 min(endPosition - startPosition.position, maxSize) //文件剩余数据的大小和最大索取大小哪个小，取最小的
 }
  }
FetchDataInfo(offsetMetadata, log.read(startPosition.position, length))

new FileMessageSet(file,
 channel,
 start = this.start + position,
 end = math.min(this.start + position + size, sizeInBytes()))

val response = new FetchResponse(fetchRequest.correlationId, dataRead.mapValues(_.data))
requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(response)))

val response = key.attachment().asInstanceOf[RequestChannel.Response]
val responseSend = response.responseSend
if(responseSend == null)
 throw new IllegalStateException("Registered for write interest but no response attached to key.")
val written = responseSend.writeTo(socketChannel)

def writeTo(destChannel: GatheringByteChannel, writePosition: Long, size: Int): Int = {
 // Ensure that the underlying size has not changed.
 val newSize = math.min(channel.size().toInt, end) - start
 if (newSize < _size.get()) {
 throw new KafkaException("Size of FileMessageSet %s has been truncated during write: old size %d, new size %d"
 .format(file.getAbsolutePath, _size.get(), newSize))
  }
 val bytesTransferred = channel.transferTo(start + writePosition, math.min(size, sizeInBytes), destChannel).toInt
  trace("FileMessageSet " + file.getAbsolutePath + " : bytes transferred : " + bytesTransferred
    + " bytes requested for transfer : " + math.min(size, sizeInBytes))
 bytesTransferred
}

val appendInfo = analyzeAndValidateMessageSet(messages)

// Check if the message sizes are valid.
val messageSize = MessageSet.entrySize(m) //单个消息的大小
if(messageSize > config.maxMessageSize) {
  BrokerTopicStats.getBrokerTopicStats(topicAndPartition.topic).bytesRejectedRate.mark(messages.sizeInBytes)
  BrokerTopicStats.getBrokerAllTopicsStats.bytesRejectedRate.mark(messages.sizeInBytes)
 throw new MessageSizeTooLargeException("Message size is %d bytes which exceeds the maximum configured message size of %d."
 .format(messageSize, config.maxMessageSize))
}

// if we just updated the current chunk and it is empty that means the fetch size is too small!
if(currentDataChunk.messages.validBytes == 0)
 throw new MessageSizeTooLargeException("Found a message larger than the maximum fetch size of this consumer on topic " +
 "%s partition %d at fetch offset %d. Increase the fetch size, or decrease the maximum message size the broker will allow."
 .format(currentDataChunk.topicInfo.topic, currentDataChunk.topicInfo.partitionId, currentDataChunk.fetchOffset))

 private def shallowValidBytes: Int = {
 if(shallowValidByteCount < 0) {
 var bytes = 0
 val iter = this.internalIterator(true)
 while(iter.hasNext) {
 val messageAndOffset = iter.next
        bytes += MessageSet.entrySize(messageAndOffset.message)
      }
//      bytes=0，代表消息不完整，当时第一次迭代就读取的消息不完整的话就会出现bytes为零，这时就需要调整消费者的消息大小了。
 this.shallowValidByteCount = bytes
    }
 shallowValidByteCount
 }