kafka源碼

『壹』 《ApacheKafka源碼剖析》pdf下載在線閱讀，求百度網盤雲資源

《Apache Kafka源碼剖析》（徐郡明）電子書網盤下載免費在線閱讀

資源鏈接：

鏈接：

書名：Apache Kafka源碼剖析

作者：徐郡明

豆瓣評分：8.4

出版社：電子工業出版社

出版年份：2017-5

頁數：604

內容簡介：

《Apache Kafka源碼剖析》以Kafka 0.10.0版本源碼為基礎，針對Kafka的架構設計到實現細節進行詳細闡述。《Apache Kafka源碼剖析》共5章，從Kafka的應用場景、源碼環境搭建開始逐步深入，不僅介紹Kafka的核心概念，而且對Kafka生產者、消費者、服務端的源碼進行深入的剖析，最後介紹Kafka常用的管理腳本實現，讓讀者不僅從宏觀設計上了解Kafka，而且能夠深入到Kafka的細節設計之中。在源碼分析的過程中，還穿插了筆者工作積累的經驗和對Kafka設計的理解，希望讀者可以舉一反三，不僅知其然，而且知其所以然。

《Apache Kafka源碼剖析》旨在為讀者閱讀Kafka源碼提供幫助和指導，讓讀者更加深入地了解Kafka的運行原理、設計理念，讓讀者在設計分布式系統時可以參考Kafka的優秀設計。《Apache Kafka源碼剖析》的內容對於讀者全面提升自己的技術能力有很大幫助。

『貳』 如何查看kafka消費者信息

在Kafak中國社區的qq群中，這個問題被提及的比例是相當高的，這也是Kafka用戶最常碰到的問題之一。本文結合Kafka源碼試圖對該問題相關的因素進行探討。希望對大家有所幫助。怎麼確定分區數？「我應該選擇幾個分區？」——如果你在Kafka中國社區的群里，這樣的問題你會經常碰到的。不過有些遺憾的是，我們似乎並沒有很權威的答案能夠解答這樣的問題。其實這也不奇怪，畢竟這樣的問題通常都是沒有固定答案的。Kafka官網上標榜自己是"high-"，即一個高吞吐量的分布式消息引擎。那麼怎麼達到高吞吐量呢？Kafka在底層摒棄了java堆緩存機制，採用了操作系統級別的頁緩存，同時將隨機寫操作改為順序寫，再結合Zero-Copy的特性極大地改善了IO性能。但是，這只是一個方面，畢竟單機優化的能力是有上限的。如何通過水平擴展甚至是線性擴展來進一步提升吞吐量呢？Kafka就是使用了分區partition，通過將topic的消息打散到多個分區並分布保存在不同的broker上實現了消息處理不管是procer還是consumer的高吞吐量。Kafka的生產者和消費者都可以多線程地並行操作，而每個線程處理的是一個分區的數據。因此分區實際上是調優Kafka並行度的最小單元。對於procer而言，它實際上是用多個線程並發地向不同分區所在的broker發起Socket連接同時給這些分區發送消息；而consumer呢，同一個消費組內的所有consumer線程都被指定topic的某一個分區進行消費具體如何確定consumer線程數目我們後面會詳細說明。所以說，如果一個topic分區越多，理論上整個集群所能達到的吞吐量就越大。但分區是否越多越好呢？顯然也不是，因為每個分區都有自己的開銷：一、客戶端/伺服器端需要使用的內存就越多先說說客戶端的情況。Kafka082之後推出了Java版的全新的procer，這個procer有個參數batchsize，默認是16KB。它會為每個分區緩存消息，一旦滿了就打包將消息批量發出。看上去這是個能夠提升性能的設計。不過很顯然，因為這個參數是分區級別的，如果分區數越多，這部分緩存所需的內存佔用也會。假設你有10000個分區，按照默認設置，這部分緩存需要佔用約157MB的內存。而consumer端呢？我們拋開獲取數據所需的內存不說，只說線程的開銷。如果還是假設有10000個分區，同時consumer線程數要匹配分區數大部分情況下是最佳的消費吞吐量配置的話，那麼在consumerclient就要創建10000個線程，也需要創建大約10000個Socket去獲取分區數據。這裡面的線程切換的開銷本身已經不容小覷了。伺服器端的開銷也不小，如果閱讀Kafka源碼的話可以發現，伺服器端的很多組件都在內存中維護了分區級別的緩存，比如controller，FetcherManager等，因此分區數越多，這種緩存的成本越久越大。二、文件句柄的開銷每個分區在底層文件系統都有屬於自己的一個目錄。該目錄下通常會有兩個文件：base_offsetlog和base_offsetindex。Kafak的controller和ReplicaManager會為每個broker都保存這兩個文件句柄filehandler。很明顯，如果分區數越多，所需要保持打開狀態的文件句柄數也就越多，最終可能會突破你的ulimit-n的限制。三、降低高可用性Kafka通過副本replica機制來保證高可用。具體做法就是為每個分區保存若干個副本replica_factor指定副本數。每個副本保存在不同的broker上。期中的一個副本充當leader副本，負責處理procer和consumer請求。其他副本充當follower角色，由Kafkacontroller負責保證與leader的同步。如果leader所在的broker掛掉了，contorller會檢測到然後在zookeeper的幫助下重選出新的leader——這中間會有短暫的不可用時間窗口，雖然大部分情況下可能只是幾毫秒級別。但如果你有10000個分區，10個broker，也就是說平均每個broker上有1000個分區。此時這個broker掛掉了，那麼zookeeper和controller需要立即對這1000個分區進行leader選舉。比起很少的分區leader選舉而言，這必然要花更長的時間，並且通常不是線性累加的。如果這個broker還同時是controller情況就更糟了。說了這么多「廢話」，很多人肯定已經不耐煩了。那你說到底要怎麼確定分區數呢？答案就是：視情況而定。基本上你還是需要通過一系列實驗和測試來確定。當然測試的依據應該是吞吐量。雖然LinkedIn這篇文章做了Kafka的基準測試，但它的結果其實對你意義不大，因為不同的硬體、軟體、負載情況測試出來的結果必然不一樣。我經常碰到的問題類似於，官網說每秒能到10MB，為什麼我的procer每秒才1MB？——且不說硬體條件，最後發現他使用的消息體有1KB，而官網的基準測試是用100B測出來的，因此根本沒有可比性。不過你依然可以遵循一定的步驟來嘗試確定分區數：創建一個只有1個分區的topic，然後測試這個topic的procer吞吐量和consumer吞吐量。假設它們的值分別是Tp和Tc，單位可以是MB/s。然後假設總的目標吞吐量是Tt，那麼分區數=Tt/maxTp,TcTp表示procer的吞吐量。測試procer通常是很容易的，因為它的邏輯非常簡單，就是直接發送消息到Kafka就好了。Tc表示consumer的吞吐量。測試Tc通常與應用的關系更大，因為Tc的值取決於你拿到消息之後執行什麼操作，因此Tc的測試通常也要麻煩一些。另外，Kafka並不能真正地做到線性擴展其實任何系統都不能，所以你在規劃你的分區數的時候最好多規劃一下，這樣未來擴展時候也更加方便。消息-分區的分配默認情況下，Kafka根據傳遞消息的key來進行分區的分配，即hashkey

『叄』 如何在windows下查看kafka源碼

Kafka is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.(Kafka是一個分布式的、可分區的(partitioned)、基於備份的(replicated)和commit-log存儲的服務.。它提供了類似於messaging system的特性,但是在設計實現上完全不同)。kafka是一種高吞吐量的分布式發布訂閱消息系統，它有如下特性：（1）、通過O(1)的磁碟數據結構提供消息的持久化，這種結構對於即使數以TB的消息存儲也能夠保持長時間的穩定性能。
（2）、高吞吐量：即使是非常普通的硬體kafka也可以支持每秒數十萬的消息。
（3）、支持通過kafka伺服器和消費機集群來分區消息。
（4）、支持Hadoop並行數據載入。
一、用Kafka裡面自帶的腳本進行編譯
下載好了Kafka源碼，裡面自帶了一個gradlew的腳本，我們可以利用這個編譯Kafka源碼：

『肆』 如何確定Kafka的分區數，key和consumer線程數

一、客戶端/伺服器端需要使用的內存就越多
先說說客戶端的情況。Kafka 0.8.2之後推出了Java版的全新的procer，這個procer有個參數batch.size，默認是16KB。它會為每個分區緩存消息，一旦滿了就打包將消息批量發出。看上去這是個能夠提升性能的設計。不過很顯然，因為這個參數是分區級別的，如果分區數越多，這部分緩存所需的內存佔用也會更多。假設你有10000個分區，按照默認設置，這部分緩存需要佔用約157MB的內存。而consumer端呢？我們拋開獲取數據所需的內存不說，只說線程的開銷。如果還是假設有10000個分區，同時consumer線程數要匹配分區數(大部分情況下是最佳的消費吞吐量配置)的話，那麼在consumer client就要創建10000個線程，也需要創建大約10000個Socket去獲取分區數據。這裡面的線程切換的開銷本身已經不容小覷了。
伺服器端的開銷也不小，如果閱讀Kafka源碼的話可以發現，伺服器端的很多組件都在內存中維護了分區級別的緩存，比如controller，FetcherManager等，因此分區數越多，這種緩存的成本越久越大。
二、文件句柄的開銷
每個分區在底層文件系統都有屬於自己的一個目錄。該目錄下通常會有兩個文件： base_offset.log和base_offset.index。Kafak的controller和ReplicaManager會為每個broker都保存這兩個文件句柄(file handler)。很明顯，如果分區數越多，所需要保持打開狀態的文件句柄數也就越多，最終可能會突破你的ulimit -n的限制。
三、降低高可用性
Kafka通過副本(replica)機制來保證高可用。具體做法就是為每個分區保存若干個副本(replica_factor指定副本數)。每個副本保存在不同的broker上。期中的一個副本充當leader 副本，負責處理procer和consumer請求。其他副本充當follower角色，由Kafka controller負責保證與leader的同步。如果leader所在的broker掛掉了，contorller會檢測到然後在zookeeper的幫助下重選出新的leader——這中間會有短暫的不可用時間窗口，雖然大部分情況下可能只是幾毫秒級別。但如果你有10000個分區，10個broker，也就是說平均每個broker上有1000個分區。此時這個broker掛掉了，那麼zookeeper和controller需要立即對這1000個分區進行leader選舉。比起很少的分區leader選舉而言，這必然要花更長的時間，並且通常不是線性累加的。如果這個broker還同時是controller情況就更糟了。
說了這么多「廢話」，很多人肯定已經不耐煩了。那你說到底要怎麼確定分區數呢？答案就是：視情況而定。基本上你還是需要通過一系列實驗和測試來確定。當然測試的依據應該是吞吐量。雖然LinkedIn這篇文章做了Kafka的基準測試，但它的結果其實對你意義不大，因為不同的硬體、軟體、負載情況測試出來的結果必然不一樣。我經常碰到的問題類似於，官網說每秒能到10MB，為什麼我的procer每秒才1MB？ —— 且不說硬體條件，最後發現他使用的消息體有1KB，而官網的基準測試是用100B測出來的，因此根本沒有可比性。不過你依然可以遵循一定的步驟來嘗試確定分區數：創建一個只有1個分區的topic，然後測試這個topic的procer吞吐量和consumer吞吐量。假設它們的值分別是Tp和Tc，單位可以是MB/s。然後假設總的目標吞吐量是Tt，那麼分區數 = Tt / max(Tp, Tc)
Tp表示procer的吞吐量。測試procer通常是很容易的，因為它的邏輯非常簡單，就是直接發送消息到Kafka就好了。Tc表示consumer的吞吐量。測試Tc通常與應用的關系更大， 因為Tc的值取決於你拿到消息之後執行什麼操作，因此Tc的測試通常也要麻煩一些。
另外，Kafka並不能真正地做到線性擴展(其實任何系統都不能)，所以你在規劃你的分區數的時候最好多規劃一下，這樣未來擴展時候也更加方便。
消息-分區的分配
默認情況下，Kafka根據傳遞消息的key來進行分區的分配，即hash(key) % numPartitions，如下圖所示:
def partition(key: Any, numPartitions: Int): Int = {
Utils.abs(key.hashCode) % numPartitions
}
這就保證了相同key的消息一定會被路由到相同的分區。如果你沒有指定key，那麼Kafka是如何確定這條消息去往哪個分區的呢？
復制代碼
if(key == null) { // 如果沒有指定key
val id = sendPartitionPerTopicCache.get(topic) // 先看看Kafka有沒有緩存的現成的分區Id
id match {
case Some(partitionId) =>
partitionId // 如果有的話直接使用這個分區Id就好了
case None => // 如果沒有的話，
val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) //找出所有可用分區的leader所在的broker
if (availablePartitions.isEmpty)
throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)
val index = Utils.abs(Random.nextInt) % availablePartitions.size // 從中隨機挑一個
val partitionId = availablePartitions(index).partitionId
sendPartitionPerTopicCache.put(topic, partitionId) // 更新緩存以備下一次直接使用
partitionId
}
}
復制代碼
可以看出，Kafka幾乎就是隨機找一個分區發送無key的消息，然後把這個分區號加入到緩存中以備後面直接使用——當然了，Kafka本身也會清空該緩存（默認每10分鍾或每次請求topic元數據時）
如何設定consumer線程數
我個人的觀點，如果你的分區數是N，那麼最好線程數也保持為N，這樣通常能夠達到最大的吞吐量。超過N的配置只是浪費系統資源，因為多出的線程不會被分配到任何分區。讓我們來看看具體Kafka是如何分配的。
topic下的一個分區只能被同一個consumer group下的一個consumer線程來消費，但反之並不成立，即一個consumer線程可以消費多個分區的數據，比如Kafka提供的ConsoleConsumer，默認就只是一個線程來消費所有分區的數據。——其實ConsoleConsumer可以使用通配符的功能實現同時消費多個topic數據，但這和本文無關。
再討論分配策略之前，先說說KafkaStream——它是consumer的關鍵類，提供了遍歷方法用於consumer程序調用實現數據的消費。其底層維護了一個阻塞隊列，所以在沒有新消息到來時，consumer是處於阻塞狀態的，表現出來的狀態就是consumer程序一直在等待新消息的到來。——你當然可以配置成帶超時的consumer，具體參看參數consumer.timeout.ms的用法。
下面說說Kafka提供的兩種分配策略： range和roundrobin，由參數partition.assignment.strategy指定，默認是range策略。本文只討論range策略。所謂的range其實就是按照階段平均分配。舉個例子就明白了，假設你有10個分區，P0 ~ P9，consumer線程數是3， C0 ~ C2，那麼每個線程都分配哪些分區呢？
C0 消費分區 0, 1, 2, 3
C1 消費分區 4, 5, 6
C2 消費分區 7, 8, 9
具體演算法就是：
復制代碼
val nPartsPerConsumer = curPartitions.size / curConsumers.size // 每個consumer至少保證消費的分區數
val nConsumersWithExtraPart = curPartitions.size % curConsumers.size // 還剩下多少個分區需要單獨分配給開頭的線程們
...
for (consumerThreadId <- consumerThreadIdSet) { // 對於每一個consumer線程
val myConsumerPosition = curConsumers.indexOf(consumerThreadId) //算出該線程在所有線程中的位置，介於[0, n-1]
assert(myConsumerPosition >= 0)
// startPart 就是這個線程要消費的起始分區數
val startPart = nPartsPerConsumer * myConsumerPosition + myConsumerPosition.min(nConsumersWithExtraPart)
// nParts 就是這個線程總共要消費多少個分區
val nParts = nPartsPerConsumer + (if (myConsumerPosition + 1 > nConsumersWithExtraPart) 0 else 1)
...
}
復制代碼
針對於這個例子，nPartsPerConsumer就是10/3=3，nConsumersWithExtraPart為10%3=1，說明每個線程至少保證3個分區，還剩下1個分區需要單獨分配給開頭的若干個線程。這就是為什麼C0消費4個分區，後面的2個線程每個消費3個分區，具體過程詳見下面的Debug截圖信息：
ctx.myTopicThreadIds
nPartsPerConsumer = 10 / 3 = 3
nConsumersWithExtraPart = 10 % 3 = 1
第一次：
myConsumerPosition = 1
startPart = 1 * 3 + min(1, 1) = 4 ---也就是從分區4開始讀
nParts = 3 + (if (1 + 1 > 1) 0 else 1) = 3 讀取3個分區， 即4,5,6
第二次：
myConsumerPosition = 0
startPart = 3 * 0 + min(1, 0) =0 --- 從分區0開始讀
nParts = 3 + (if (0 + 1 > 1) 0 else 1) = 4 讀取4個分區，即0,1,2,3
第三次：
myConsumerPosition = 2
startPart = 3 * 2 + min(2, 1) = 7 --- 從分區7開始讀
nParts = 3 + if (2 + 1 > 1) 0 else 1) = 3 讀取3個分區，即7, 8, 9
至此10個分區都已經分配完畢
說到這里，經常有個需求就是我想讓某個consumer線程消費指定的分區而不消費其他的分區。坦率來說，目前Kafka並沒有提供自定義分配策略。做到這點很難，但仔細想一想，也許我們期望Kafka做的事情太多了，畢竟它只是個消息引擎，在Kafka中加入消息消費的邏輯也許並不是Kafka該做的事情。

『伍』 學習apache kafka源碼剖析需要什麼基礎

先搞清楚STL怎麼用並大量使用相當長的時間，代碼風格盡量STL化（這個真是看STL源碼的前提，我就是受不了全是模板和迭代器的代碼，所以至今沒去研究STL源碼）

還有，現在對「基礎較好」、「熟練」、「精通」之類的詞本能的不信任

『陸』 apache kafka何時開源

良好的擴展性和性能優勢.不過到目前為止,我們應該很清楚認識到,kafka並沒有提供JMS中的"事務性""消息傳輸擔保(消息確認機制)""消息分組"

『柒』 基於librdkafka庫寫的源代碼怎麼編譯

可以放在當前目錄下，但是要設置一下庫文件的路徑：LD_LIBRARY_PATH=./:/usr/local/pet20/lib:/lib/:/usr/local/lib export LD_LIBRARY_PATH 這樣，在調用的時候就會自動從當前目錄找。 如果是顯式調用則不用，只要在程序里指定.so的文件路徑就可以了。所以放在當前目錄下也是沒問題的。

『捌』 kafka 有java的源碼嗎

我這里是使用的是，kafka自帶的zookeeper。
以及關於kafka的日誌文件啊，都放在默認里即/tmp下，我沒修改。保存默認的

1、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ jps
2625 Jps
2、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ bin/zookeeper-server-start.sh config/zookeeper.properties
此刻，這時，會一直停在這，因為是前端運行。
另開一窗口，
3、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ bin/kafka-server-start.sh config/server.properties
也是前端運行。

『玖』 如何保證kafka 的消息機制 ack-fail 源碼跟蹤

Kafka is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.(Kafka布式、區(partitioned)、基於備份(replicated)commit-log存儲服務.提供類似於messaging system特性,設計實現完全同)kafka種高吞吐量布式發布訂閱消息系統特性：
（1）、通O(1)磁碟數據結構提供消息持久化種結構於即使數TB消息存儲能夠保持間穩定性能
（2）、高吞吐量：即使非普通硬體kafka支持每秒數十萬消息
（3）、支持通kafka伺服器消費機集群區消息
（4）、支持Hadoop並行數據載入
、用Kafka面自帶腳本進行編譯
載Kafka源碼面自帶gradlew腳本我利用編譯Kafka源碼：
1 # wget
2 # tar -zxf kafka-0.8.1.1-src.tgz
3 # cd kafka-0.8.1.1-src
4 # ./gradlew releaseTarGz
運行面命令進行編譯現異信息：
01 :core:signArchives FAILED
02
03 FAILURE: Build failed with an exception.
04
05 * What went wrong:
06 Execution failed for task ':core:signArchives'.
07 > Cannot perform signing task ':core:signArchives' because it
08 has no configured signatory
09
10 * Try:
11 Run with --stacktrace option to get the stack trace. Run with
12 --info or --debug option to get more log output.
13
14 BUILD FAILED
bug（）用面命令進行編譯
1 ./gradlew releaseTarGzAll -x signArchives
候編譯功（編譯程現）編譯程我指定應Scala版本進行編譯：
1 ./gradlew -PscalaVersion=2.10.3 releaseTarGz -x signArchives
編譯完core/build/distributions/面kafka_2.10-0.8.1.1.tgz文件網載直接用
二、利用sbt進行編譯
我同用sbt編譯Kafka步驟：
01 # git clone
02 # cd kafka
03 # git checkout -b 0.8 remotes/origin/0.8
04 # ./sbt update
05 [info] [SUCCESSFUL ] org.eclipse.jdt#core;3.1.1!core.jar (2243ms)
06 [info] downloading ...
07 [info] [SUCCESSFUL ] ant#ant;1.6.5!ant.jar (1150ms)
08 [info] Done updating.
09 [info] Resolving org.apache.hadoop#hadoop-core;0.20.2 ...
10 [info] Done updating.
11 [info] Resolving com.yammer.metrics#metrics-annotation;2.2.0 ...
12 [info] Done updating.
13 [info] Resolving com.yammer.metrics#metrics-annotation;2.2.0 ...
14 [info] Done updating.
15 [success] Total time: 168 s, completed Jun 18, 2014 6:51:38 PM
16
17 # ./sbt package
18 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
19 Getting Scala 2.8.0 ...
20 :: retrieving :: org.scala-sbt#boot-scala
21 confs: [default]
22 3 artifacts copied, 0 already retrieved (14544kB/27ms)
23 [success] Total time: 1 s, completed Jun 18, 2014 6:52:37 PM
於Kafka 0.8及版本需要運行命令：
01 # ./sbt assembly-package-dependency
02 [info] Loading project definition from /export1/spark/kafka/project
03 [warn] Multiple resolvers having different access mechanism configured with
04 same name 'sbt-plugin-releases'. To avoid conflict, Remove plicate project
05 resolvers (`resolvers`) or rename publishing resolver (`publishTo`).
06 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
07 [warn] Credentials file /home/wyp/.m2/.credentials does not exist
08 [info] Including slf4j-api-1.7.2.jar
09 [info] Including metrics-annotation-2.2.0.jar
10 [info] Including scala-compiler.jar
11 [info] Including scala-library.jar
12 [info] Including slf4j-simple-1.6.4.jar
13 [info] Including metrics-core-2.2.0.jar
14 [info] Including snappy-java-1.0.4.1.jar
15 [info] Including zookeeper-3.3.4.jar
16 [info] Including log4j-1.2.15.jar
17 [info] Including zkclient-0.3.jar
18 [info] Including jopt-simple-3.2.jar
19 [warn] Merging 'META-INF/NOTICE' with strategy 'rename'
20 [warn] Merging 'org/xerial/snappy/native/README' with strategy 'rename'
21 [warn] Merging 'META-INF/maven/org.xerial.snappy/snappy-java/LICENSE'
22 with strategy 'rename'
23 [warn] Merging 'LICENSE.txt' with strategy 'rename'
24 [warn] Merging 'META-INF/LICENSE' with strategy 'rename'
25 [warn] Merging 'META-INF/MANIFEST.MF' with strategy 'discard'
26 [warn] Strategy 'discard' was applied to a file
27 [warn] Strategy 'rename' was applied to 5 files
28 [success] Total time: 3 s, completed Jun 18, 2014 6:53:41 PM
我sbt面指定scala版本：
01 <!--
02 User: 往記憶
03 Date: 14-6-18
04 Time: 20:20
05 bolg:
06 本文址：/archives/1044
07 往記憶博客專注於hadoop、hive、spark、shark、flume技術博客量干貨
08 往記憶博客微信公共帳號：iteblog_hadoop
09 -->
10 sbt "++2.10.3 update"
11 sbt "++2.10.3 package"
12 sbt "++2.10.3 assembly-package-dependency"

『拾』 如何刪除kafka積壓數據

Kafka刪除數據有兩種方式

• 按照時間，超過一段時間後刪除過期消息

• 按照消息大小，消息數量超過一定大小後刪除最舊的數據

Kafka刪除數據的最小單位：segment

Kafka刪除數據主邏輯：kafka源碼



defcleanupLogs() {debug("Beginning log cleanup...")vartotal=0valstartMs=time.millisecondsfor(log <- allLogs;if!log.config.compact) {debug("Garbage collecting '"+ log.name +"'")total +=cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)}debug("Log cleanup completed. "+ total +" files deleted in "+(time.milliseconds - startMs) /1000+" seconds")}

Kafka一段時間（配置文件設置）調用一次 cleanupLogs，刪除所有應該刪除的日誌數據。

cleanupExpiredSegments 負責清理超時的數據



(log:Log):Int={valstartMs=time.millisecondslog.deleteOldSegments(startMs -_.lastModified > log.config.retentionMs)}

cleanupSegmentsToMaintainSize 負責清理超過大小的數據



<br>(log:Log):Int={if(log.config.retentionSize <0|| log.size < log.config.retentionSize)return0vardiff=log.size - log.config.retentionSizedefshouldDelete(segment:LogSegment)={if(diff - segment.size >=0) {diff -=segment.sizetrue}else{false}}log.deleteOldSegments(shouldDelete)}