前言

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

• NIO
  
• Zero Copy
  
• 磁盤順序讀寫
  
• Queue數據結構的極致使用
  

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

• Kafka在什么場景下用了這個技術
  
• Zero-Copy 是如何被調用，并且發揮作用的。
  

Kafka在什么場景下使用該技術

答案是: 消息消費的時候? 包括外部Consumer以及Follower 從partiton Leader同步數據，都是如此。簡單描述就是： Consumer從Broker獲取文件數據的時候，直接通過下面的方法進行channel到channel的數據傳輸。 java.nio.FileChannel.transferTo( long position, long count, WritableByteChannel target)` 也就是說你的數據源是一個Channel,數據接收端也是一個Channel(SocketChannel),則通過該方式進行數據傳輸，是直接在內核態進行的，避免拷貝數據導致的內核態和用戶態的多次切換。

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

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

數據的生成

對應的類名稱為： kaka.server.KafkaApis 該類是負責真正的Kafka業務邏輯處理的。在此之前的，譬如 SocketServer等類似Tomcat服務器一樣，側重于交互，屬于框架層次的東西。KafkaApis 則類似于部署在Tomcat里的應用。 def handle(request: RequestChannel.Request) {ApiKeys.forId(request.requestId) match {case ApiKeys.PRODUCE => handleProducerRequest(request)case ApiKeys.FETCH => handleFetchRequest(request)..... handle 方法是所有處理的入口，然后根據請求的不同，有不同的處理邏輯。這里我們關注ApiKeys.FETCH這塊，也就是有消費者要獲取數據的邏輯。進入 handleFetchRequest方法，你會看到最后一行代碼如下： replicaManager.fetchMessages( fetchRequest.maxWait.toLong, fetchRequest.replicaId, fetchRequest.minBytes, authorizedRequestInfo, sendResponseCallback) ReplicaManager 包含所有主題的所有partition消息。大部分針對Partition的操作都是通過該類來完成的。 replicaManager.fetchMessages 這個方法非常的長。我們只關注一句代碼：
val logReadResults = readFromLocalLog(fetchOnlyFromLeader, fetchOnlyCommitted, fetchInfo) 該方法獲取本地日志信息數據。內部會調用kafka.cluster.Log對象的read方法： log.read(offset, fetchSize, maxOffsetOpt) Log 對象是啥呢？其實就是對應的一個Topic的Partition. 一個Partition是由很多端(Segment)組成的，這和Lucene非常相似。一個Segment就是一個文件。實際的數據自然是從這里讀到的。代碼如下： val fetchInfo = entry.getValue.read(startOffset, maxOffset, maxLength, maxPosition) 這里的fetchInfo(FetchDataInfo)對象包含兩個字段: offsetMetadata FileMessageSet FileMessageSet 其實就是用戶在這個Partition這一次消費能夠拿到的數據集合。當然，真實的數據還躺在byteBuffer里，并沒有記在到內存中。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)}).toIntbytesTransferred} 這里我們看到了久違的transferFrom方法。那么這個方法什么時候被調用呢？我們先擱置下，因為那個是另外一個流程。我們繼續分析上面的代碼。也就是接著從這段代碼開始分析： val logReadResults = readFromLocalLog(fetchOnlyFromLeader, fetchOnlyCommitted, fetchInfo) 獲取到這個信息后，會執行如下操作： val fetchPartitionData = logReadResults.mapValues(result => FetchResponsePartitionData(result.errorCode, result.hw, result.info.messageSet)) responseCallback(fetchPartitionData) logReadResults 的信息被包裝成FetchResponsePartitionData, FetchResponsePartitionData 包喊了我們的FileMessageSet 對象。還記得么，這個對象包含了我們要跟蹤的tranferTo方法。然后FetchResponsePartitionData 會給responseCallback作為參數進行回調。 responseCallback 的函數簽名如下(我去掉了一些我們不關心的信息)： def sendResponseCallback(responsePartitionData: Map[TopicAndPartition, FetchResponsePartitionData]) {val mergedResponseStatus = responsePartitionData ++ unauthorizedResponseStatusdef fetchResponseCallback(delayTimeMs: Int) {val response = FetchResponse(fetchRequest.correlationId, mergedResponseStatus, fetchRequest.versionId, delayTimeMs)requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(request.connectionId, response)))}} 我們重點關注這個回調方法里的fetchResponseCallback。 我們會發現這里 FetchResponsePartitionData 會被封裝成一個FetchResponseSend ,然后由requestChannel發送出去。 因為Kafka完全應用是NIO的異步機制，所以到這里，我們無法再跟進去了，需要從另外一部分開始分析。

數據的發送

前面只是涉及到數據的獲取。讀取日志，并且獲得對應MessageSet對象。MessageSet 是一段數據的集合，但是該數據沒有真實的被加載。這里會涉及到Kafka 如何將數據發送回Consumer端。 在SocketServer，也就是負責和所有的消費者打交道，建立連接的中樞里，會不斷的進行poll操作 override def run() {startupComplete()while(isRunning) {try {// setup any new connections that have been queued upconfigureNewConnections()// register any new responses for writingprocessNewResponses() 首先會注冊新的連接，如果有的話。接著就是處理新的響應了。還記得剛剛上面我們通過requestChannel把FetchResponseSend發出來吧。 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.scalapublic void setSend(Send send) {this.send = send; this.transportLayer.addInterestOps(SelectionKey.OP_WRITE); } 為了方便看代碼，我對代碼做了改寫。我們看到，其實send就是做了一個WRITE時間注冊。這個是和NIO機制相關的。如果大家看的有障礙，不妨先學習下相關的機制。 回到 SocketServer 的run方法里，也就是上面已經貼過的代碼： override def run() {startupComplete()while(isRunning) {try {// setup any new connections that have been queued upconfigureNewConnections()// register any new responses for writingprocessNewResponses()try {selector.poll(300)} catch {case...} SocketServer 會poll隊列，一旦對應的KafkaChannel 寫操作ready了，就會調用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();} 依然的，為了減少代碼，我做了些調整，其中write會調用 send方法，對應的Send對象其實就是上面我們注冊的FetchResponseSend 對象。 這段代碼里真實發送數據的代碼是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)....} 這里我依然做了代碼簡化，只讓我們關注核心的。 這里最后是調用了sends的writeTo方法，而sends 其實是個MultiSend。這個MultiSend 里有兩個東西：

• topicAndPartition.partition: 分區
  
• message:FetchResponsePartitionData ?
  

還記得這個FetchResponsePartitionData ?么？我們的MessageSet 就被放在了FetchResponsePartitionData這個對象里。 TopicDataSend 也包含了sends,該sends 包含了 PartitionDataSend，而 PartitionDataSend則包含了FetchResponsePartitionData。 最后進行writeTo的時候，其實是調用了 //partitionData 就是 FetchResponsePartitionData //messages 其實就是FileMessageSet val bytesSent = partitionData.messages.writeTo(channel, messagesSentSize, messageSize - messagesSentSize) 如果你還記得的話，FileMessageSet 也有個writeTo方法，就是我們之前已經提到過的那段代碼： 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)}).toIntbytesTransferred} 終于走到最底層了，最后其實是通過tl.transferFrom(channel, position, count) 來完成最后的數據發送的。這里你可能比較好奇，不應該是調用transferTo 方法么? transferFrom其實是Kafka自己封裝的一個方法，最終里面調用的也是transerTo: @Overridepublic long transferFrom(FileChannel fileChannel, long position, long count) throws IOException {return fileChannel.transferTo(position, count, socketChannel);}

總結

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

總結

以上是生活随笔為你收集整理的Kafka Zero-Copy 使用分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。