之前有聽過Zero-Copy 技術(shù)雾鬼，而Kafka是典型的使用者葱轩。網(wǎng)上找了找，竟然沒有找到合適的介紹文章狠毯。正好這段時間正在閱讀Kafka的相關(guān)代碼护糖，于是有了這篇內(nèi)容。這篇文章會簡要介紹Zero-Copy技術(shù)在Kafka的使用情況嚼松，希望能給大家一定借鑒和學(xué)習(xí)樣例嫡良。
前言
Kafka 我個人感覺是性能優(yōu)化的典范。而且使用Scala開發(fā)献酗，代碼寫的也很漂亮的皆刺。重點我覺得有四個
- NIO
- Zero Copy
- 磁盤順序讀寫
- Queue數(shù)據(jù)結(jié)構(gòu)的極致使用
Zero-Copy 實際的原理，大家還是去Google下凌摄。這篇文章重點會分析這項技術(shù)是怎么被嵌入到Kafa里的。包含兩部分：
- Kafka在什么場景下用了這個技術(shù)
- 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ù)導(dǎo)致的內(nèi)核態(tài)和用戶態(tài)的多次切換。
Kafka 如何使用Zero-Copy流程分析
估計看完這段內(nèi)容施戴，你對整個Kafka的數(shù)據(jù)處理流程也差不多了解了個大概反浓。為了避免過于繁雜，以至于將整個Kafka的體系都拖進來赞哗，我們起始點從KafkaApis相關(guān)的類開始雷则。
數(shù)據(jù)的生成
對應(yīng)的類名稱為：
kaka.server.KafkaApis
該類是負責(zé)真正的Kafka業(yè)務(wù)邏輯處理的。在此之前的肪笋，譬如 SocketServer等類似Tomcat服務(wù)器一樣月劈，側(cè)重于交互度迂，屬于框架層次的東西。KafkaApis 則類似于部署在Tomcat里的應(yīng)用猜揪。
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 對象是啥呢？其實就是對應(yīng)的一個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ù)集合蟹略。當(dāng)然，真實的數(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完全應(yīng)用是NIO的異步機制，所以到這里捶障，我們無法再跟進去了僧须，需要從另外一部分開始分析。
數(shù)據(jù)的發(fā)送
前面只是涉及到數(shù)據(jù)的獲取项炼。讀取日志担平，并且獲得對應(yīng)MessageSet對象示绊。MessageSet 是一段數(shù)據(jù)的集合，但是該數(shù)據(jù)沒有真實的被加載暂论。
這里會涉及到Kafka 如何將數(shù)據(jù)發(fā)送回Consumer端面褐。
在SocketServer，也就是負責(zé)和所有的消費者打交道取胎，建立連接的中樞里展哭，會不斷的進行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()
首先會注冊新的連接，如果有的話闻蛀。接著就是處理新的響應(yīng)了匪傍。還記得剛剛上面我們通過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)的俐芯。如果大家看的有障礙棵介，不妨先學(xué)習(xí)下相關(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隊列邮辽，一旦對應(yīng)的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方法莽使，對應(yīng)的Send對象其實就是上面我們注冊的FetchResponseSend 對象。
這段代碼里真實發(fā)送數(shù)據(jù)的代碼是send.writeTo(transportLayer);笙僚，
對應(yīng)的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ā)送的燃领。這里你可能比較好奇士聪，不應(yīng)該是調(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機制区岗，我們在分析的過程中，為了方便毁枯，把這塊完全給抹掉了慈缔。
