之前有聽過Zero-Copy 技術(shù)，而Kafka是典型的使用者。網(wǎng)上找了找，竟然沒有找到合適的介紹文章。正好這段時間正在閱讀Kafka的相關(guān)代碼，于是有了這篇內(nèi)容。這篇文章會簡要介紹Zero-Copy技術(shù)在Kafka的使用情況，希望能給大家一定借鑒和學習樣例。

前言

Kafka 我個人感覺是性能優(yōu)化的典范。而且使用Scala開發(fā)，代碼寫的也很漂亮的。重點我覺得有四個

• NIO
• Zero Copy
• 磁盤順序讀寫
• Queue數(shù)據(jù)結(jié)構(gòu)的極致使用

Zero-Copy 實際的原理，大家還是去Google下。這篇文章重點會分析這項技術(shù)是怎么被嵌入到Kafa里的。包含兩部分：

1. Kafka在什么場景下用了這個技術(shù)
2. Zero-Copy 是如何被調(diào)用，并且發(fā)揮作用的。

Kafka在什么場景下使用該技術(shù)

答案是:

消息消費的時候

包括外部Consumer以及Follower 從partiton Leader同步數(shù)據(jù)，都是如此。簡單描述就是：

Consumer從Broker獲取文件數(shù)據(jù)的時候，直接通過下面的方法進行channel到channel的數(shù)據(jù)傳輸。

java.nio.FileChannel.transferTo(
long position, 
long count,                                
WritableByteChannel target)`

也就是說你的數(shù)據(jù)源是一個Channel,數(shù)據(jù)接收端也是一個Channel(SocketChannel),則通過該方式進行數(shù)據(jù)傳輸，是直接在內(nèi)核態(tài)進行的，避免拷貝數(shù)據(jù)導致的內(nèi)核態(tài)和用戶態(tài)的多次切換。

Kafka 如何使用Zero-Copy流程分析

估計看完這段內(nèi)容，你對整個Kafka的數(shù)據(jù)處理流程也差不多了解了個大概。為了避免過于繁雜，以至于將整個Kafka的體系都拖進來，我們起始點從KafkaApis相關(guān)的類開始。

數(shù)據(jù)的生成

對應的類名稱為：

kaka.server.KafkaApis

該類是負責真正的Kafka業(yè)務邏輯處理的。在此之前的，譬如 SocketServer等類似Tomcat服務器一樣，側(cè)重于交互，屬于框架層次的東西。KafkaApis 則類似于部署在Tomcat里的應用。

def handle(request: RequestChannel.Request) {
       ApiKeys.forId(request.requestId) match {
        case ApiKeys.PRODUCE => handleProducerRequest(request)
        case ApiKeys.FETCH => handleFetchRequest(request)
        .....

handle 方法是所有處理的入口，然后根據(jù)請求的不同，有不同的處理邏輯。這里我們關(guān)注ApiKeys.FETCH這塊，也就是有消費者要獲取數(shù)據(jù)的邏輯。進入 handleFetchRequest方法，你會看到最后一行代碼如下：

replicaManager.fetchMessages(  
       fetchRequest.maxWait.toLong, 
      fetchRequest.replicaId, 
      fetchRequest.minBytes,  
      authorizedRequestInfo,  
      sendResponseCallback)

ReplicaManager 包含所有主題的所有partition消息。大部分針對Partition的操作都是通過該類來完成的。

replicaManager.fetchMessages 這個方法非常的長。我們只關(guān)注一句代碼：

val logReadResults = readFromLocalLog(fetchOnlyFromLeader, fetchOnlyCommitted, fetchInfo)

該方法獲取本地日志信息數(shù)據(jù)。內(nèi)部會調(diào)用kafka.cluster.Log對象的read方法：

log.read(offset, fetchSize, maxOffsetOpt)

Log 對象是啥呢？其實就是對應的一個Topic的Partition. 一個Partition是由很多端(Segment)組成的，這和Lucene非常相似。一個Segment就是一個文件。實際的數(shù)據(jù)自然是從這里讀到的。代碼如下：

val fetchInfo = entry.getValue.read(startOffset, maxOffset, maxLength, maxPosition)

這里的fetchInfo(FetchDataInfo)對象包含兩個字段:

• offsetMetadata
• FileMessageSet

FileMessageSet 其實就是用戶在這個Partition這一次消費能夠拿到的數(shù)據(jù)集合。當然，真實的數(shù)據(jù)還躺在byteBuffer里，并沒有記在到內(nèi)存中。FileMessageSet 里面包含了一個很重要的方法：

def writeTo(destChannel: GatheringByteChannel, writePosition: Long, size: Int): Int = {
    ......
    
    val bytesTransferred = (destChannel match {
      case tl: TransportLayer => tl.transferFrom(channel, position, count)
      case dc => channel.transferTo(position, count, dc)
    }).toInt
   
    bytesTransferred
  }

這里我們看到了久違的transferFrom方法。那么這個方法什么時候被調(diào)用呢？我們先擱置下，因為那個是另外一個流程。我們繼續(xù)分析上面的代碼。也就是接著從這段代碼開始分析：

val logReadResults = readFromLocalLog(fetchOnlyFromLeader, fetchOnlyCommitted, fetchInfo)

獲取到這個信息后，會執(zhí)行如下操作：

val fetchPartitionData = logReadResults.mapValues(result =>  FetchResponsePartitionData(result.errorCode, result.hw, result.info.messageSet))
responseCallback(fetchPartitionData)

logReadResults 的信息被包裝成FetchResponsePartitionData, FetchResponsePartitionData 包喊了我們的FileMessageSet 對象。還記得么，這個對象包含了我們要跟蹤的tranferTo方法。然后FetchResponsePartitionData 會給responseCallback作為參數(shù)進行回調(diào)。

responseCallback 的函數(shù)簽名如下(我去掉了一些我們不關(guān)心的信息)：

def sendResponseCallback(responsePartitionData: Map[TopicAndPartition, FetchResponsePartitionData]) {
      val mergedResponseStatus = responsePartitionData ++ unauthorizedResponseStatus

      def fetchResponseCallback(delayTimeMs: Int) {
        val response = FetchResponse(fetchRequest.correlationId, mergedResponseStatus, fetchRequest.versionId, delayTimeMs)
        requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(request.connectionId, response)))
      }

    }

我們重點關(guān)注這個回調(diào)方法里的fetchResponseCallback。 我們會發(fā)現(xiàn)這里 FetchResponsePartitionData 會被封裝成一個FetchResponseSend ,然后由requestChannel發(fā)送出去。

因為Kafka完全應用是NIO的異步機制，所以到這里，我們無法再跟進去了，需要從另外一部分開始分析。

數(shù)據(jù)的發(fā)送

前面只是涉及到數(shù)據(jù)的獲取。讀取日志，并且獲得對應MessageSet對象。MessageSet 是一段數(shù)據(jù)的集合，但是該數(shù)據(jù)沒有真實的被加載。
這里會涉及到Kafka 如何將數(shù)據(jù)發(fā)送回Consumer端。

在SocketServer，也就是負責和所有的消費者打交道，建立連接的中樞里，會不斷的進行poll操作

override def run() {
    startupComplete()
    while(isRunning) {
      try {
        // setup any new connections that have been queued up
        configureNewConnections()
        // register any new responses for writing
        processNewResponses()

首先會注冊新的連接，如果有的話。接著就是處理新的響應了。還記得剛剛上面我們通過requestChannel把FetchResponseSend發(fā)出來吧。

private def processNewResponses() {
    var curr = requestChannel.receiveResponse(id)
    while(curr != null) {
      try {
        curr.responseAction match {         
          case RequestChannel.SendAction =>
            selector.send(curr.responseSend)
            inflightResponses += (curr.request.connectionId -> curr)
          
        }
      } finally {
        curr = requestChannel.receiveResponse(id)
      }
    }
  }

這里類似的，processNewResponses方法會先通過send方法把FetchResponseSend注冊到selector上。 這個操作其實做的事情如下：

//SocketServer.scala    
public void send(Send send) {
        KafkaChannel channel = channelOrFail(send.destination());
        channel.setSend(send);
    }

//KafkaChannel.scala
   public void setSend(Send send) {
         this.send = send;          this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);     
    }

為了方便看代碼，我對代碼做了改寫。我們看到，其實send就是做了一個WRITE時間注冊。這個是和NIO機制相關(guān)的。如果大家看的有障礙，不妨先學習下相關(guān)的機制。

回到 SocketServer 的run方法里，也就是上面已經(jīng)貼過的代碼：

  override def run() {
    startupComplete()
    while(isRunning) {
      try {
        // setup any new connections that have been queued up
        configureNewConnections()
        // register any new responses for writing
        processNewResponses()

        try {
          selector.poll(300)
        } catch {
          case...
        }

SocketServer 會poll隊列，一旦對應的KafkaChannel 寫操作ready了，就會調(diào)用KafkaChannel的write方法：

//KafkaChannel.scala
public Send write() throws IOException {
        if (send != null && send(send)) 
    }
//
//KafkaChannel.scala
private boolean send(Send send) throws IOException {
        send.writeTo(transportLayer);
        if (send.completed())
            transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

        return send.completed();
    }

依然的，為了減少代碼，我做了些調(diào)整，其中write會調(diào)用 send方法，對應的Send對象其實就是上面我們注冊的FetchResponseSend 對象。

這段代碼里真實發(fā)送數(shù)據(jù)的代碼是send.writeTo(transportLayer);，

對應的writeTo方法為：

private val sends = new MultiSend(dest, JavaConversions.seqAsJavaList(fetchResponse.dataGroupedByTopic.toList.map {
    case(topic, data) => new TopicDataSend(dest, TopicData(topic,
                                                     data.map{case(topicAndPartition, message) => (topicAndPartition.partition, message)}))
    }))
override def writeTo(channel: GatheringByteChannel): Long = {
    .....    
     written += sends.writeTo(channel)
    ....
  }

這里我依然做了代碼簡化，只讓我們關(guān)注核心的。 這里最后是調(diào)用了sends的writeTo方法，而sends 其實是個MultiSend。
這個MultiSend 里有兩個東西：

• topicAndPartition.partition: 分區(qū)
• message:FetchResponsePartitionData

還記得這個FetchResponsePartitionData 么？我們的MessageSet 就被放在了FetchResponsePartitionData這個對象里。

TopicDataSend 也包含了sends,該sends 包含了 PartitionDataSend，而 PartitionDataSend則包含了FetchResponsePartitionData。

最后進行writeTo的時候，其實是調(diào)用了

//partitionData 就是 FetchResponsePartitionData
//messages 其實就是FileMessageSet
val bytesSent = partitionData.messages.writeTo(channel, messagesSentSize, messageSize - messagesSentSize)

如果你還記得的話，F(xiàn)ileMessageSet 也有個writeTo方法，就是我們之前已經(jīng)提到過的那段代碼：

def writeTo(destChannel: GatheringByteChannel, writePosition: Long, size: Int): Int = {
    ......

    val bytesTransferred = (destChannel match {
      case tl: TransportLayer => tl.transferFrom(channel, position, count)
      case dc => channel.transferTo(position, count, dc)
    }).toInt

    bytesTransferred
  }

終于走到最底層了，最后其實是通過tl.transferFrom(channel, position, count) 來完成最后的數(shù)據(jù)發(fā)送的。這里你可能比較好奇，不應該是調(diào)用transferTo 方法么? transferFrom其實是Kafka自己封裝的一個方法，最終里面調(diào)用的也是transerTo:

  @Override
    public long transferFrom(FileChannel fileChannel, long position, long count) throws IOException {
        return fileChannel.transferTo(position, count, socketChannel);
    }

總結(jié)

Kafka的整個調(diào)用棧還是非常繞的。尤其是引入了NIO的事件機制，有點類似Shuffle,把流程調(diào)用給切斷了，無法簡單通過代碼引用來進行跟蹤。Kafka還有一個非常優(yōu)秀的機制就是DelayQueue機制，我們在分析的過程中，為了方便，把這塊完全給抹掉了。