kafka架构、亿级数据如何优化GC
大家都知道Kafka是一個高吞吐的消息隊列，是大數據場景首選的消息隊列，這種場景就意味著發送單位時間消息的量會特別的大，那既然如此巨大的數據量，kafka是如何支撐起如此龐大的數據量的分發的呢？今天我們從kafka架構以如何優化GC兩個方面講解.
kafka架構
既然要說kafka是如何通過內存緩沖池設計來優化JVM的GC問題，那么，如果不清楚kafka 的架構設計，又怎么更好的調優呢？起碼的我們要知道基礎的才能往更好的出發呀，對吧？先來看一些技術名詞：
Topic：用于劃分Message的邏輯概念，一個Topic可以分布在多個Broker上。
Partition：是Kafka中橫向擴展和一切并行化的基礎，每個Topic都至少被切分為1個Partition。
Offset：消息在Partition中的編號，編號順序不跨Partition。
Consumer：用于從Broker中取出/消費Message。
Producer：用于往Broker中發送/生產Message。
Replication：Kafka支持以Partition為單位對Message進行冗余備份，每個Partition都可以配置至少1個Replication(當僅1個Replication時即僅該Partition本身)。
Leader：每個Replication集合中的Partition都會選出一個唯一的Leader，所有的讀寫請求都由Leader處理。其他Replicas從Leader處把數據更新同步到本地，過程類似大家熟悉的MySQL中的Binlog同步。
Broker：Kafka中使用Broker來接受Producer和Consumer的請求，并把Message持久化到本地磁盤。每個Cluster當中會選舉出一個Broker來擔任Controller，負責處理Partition的Leader選舉，協調Partition遷移等工作。
ISR(In-Sync Replica)：是Replicas的一個子集，表示目前Alive且與Leader能夠“Catch-up”的Replicas集合。由于讀寫都是首先落到Leader上，所以一般來說通過同步機制從Leader上拉取數據的Replica都會和Leader有一些延遲(包括了延遲時間和延遲條數兩個維度)，任意一個超過閾值都會把該Replica踢出ISR。每個Partition都有它自己獨立的ISR。
以上幾乎是我們在使用Kafka的過程中經常會遇到的名詞，同時也無一不是最核心的概念或組件，感覺到從設計本身來說，Kafka還是足夠簡潔的。這次本文圍繞Kafka優異的吞吐性能，逐個介紹一下其設計與實現當中所使用的各項“黑科技”。
Broker
不同于Redis和MemcacheQ等內存消息隊列，Kafka的設計是把所有的Message都要寫入速度低容量大的硬盤，以此來換取更強的存儲能力。實際上，Kafka使用硬盤并沒有帶來過多的性能損失，“規規矩矩”的抄了一條“近道”。
首先，說“規規矩矩”是因為Kafka在磁盤上只做Sequence I/O，由于消息系統讀寫的特殊性，這并不存在什么問題。關于磁盤I/O的性能，引用一組Kafka官方給出的測試數據(Raid-5，7200rpm)：
Sequence I/O: 600MB/s
Random I/O: 100KB/s
所以通過只做Sequence I/O的限制，規避了磁盤訪問速度低下對性能可能造成的影響。接下來我們再聊一聊Kafka是如何“抄近道的”。
首先，Kafka重度依賴底層操作系統提供的PageCache功能。當上層有寫操作時，操作系統只是將數據寫入PageCache，同時標記Page屬性為Dirty。當讀操作發生時，先從PageCache中查找，如果發生缺頁才進行磁盤調度，最終返回需要的數據。實際上PageCache是把盡可能多的空閑內存都當做了磁盤緩存來使用。同時如果有其他進程申請內存，回收PageCache的代價又很小，所以現代的OS都支持PageCache。使用PageCache功能同時可以避免在JVM內部緩存數據，JVM為我們提供了強大的GC能力，同時也引入了一些問題不適用于Kafka的設計。
? 如果在Heap內管理緩存，JVM的GC線程會頻繁掃描Heap空間，帶來不必要的開銷。如果Heap過大，執行一次Full GC對系統的可用性來說將是極大的挑戰。
? 所有在在JVM內的對象都不免帶有一個Object Overhead(千萬不可小視)，內存的有效空間利用率會因此降低。
? 所有的In-Process Cache在OS中都有一份同樣的PageCache。所以通過將緩存只放在PageCache，可以至少讓可用緩存空間翻倍。
? 如果Kafka重啟，所有的In-Process Cache都會失效，而OS管理的PageCache依然可以繼續使用。
PageCache還只是第一步，Kafka為了進一步的優化性能還采用了Sendfile技術。在解釋Sendfile之前，首先介紹一下傳統的網絡I/O操作流程，大體上分為以下4步：
OS 從硬盤把數據讀到內核區的PageCache。
用戶進程把數據從內核區Copy到用戶區。
然后用戶進程再把數據寫入到Socket，數據流入內核區的Socket Buffer上。
OS 再把數據從Buffer中Copy到網卡的Buffer上，這樣完成一次發送。
整個過程共經歷兩次Context Switch，四次System Call。同一份數據在內核Buffer與用戶Buffer之間重復拷貝，效率低下。其中2、3兩步沒有必要，完全可以直接在內核區完成數據拷貝。這也正是Sendfile所解決的問題，經過Sendfile優化后，整個I/O過程就變成了下面這個樣子。
通過以上的介紹不難看出，Kafka的設計初衷是盡一切努力在內存中完成數據交換，無論是對外作為一整個消息系統，或是內部同底層操作系統的交互。如果Producer和Consumer之間生產和消費進度上配合得當，完全可以實現數據交換零I/O。這也就是我為什么說Kafka使用“硬盤”并沒有帶來過多性能損失的原因。下面是我在生產環境中采到的一些指標：
(20 Brokers, 75 Partitions per Broker, 110k msg/s)
此時的集群只有寫，沒有讀操作。10M/s左右的Send的流量是Partition之間進行Replicate而產生的。從recv和writ的速率比較可以看出，寫盤是使用Asynchronous+Batch的方式，底層OS可能還會進行磁盤寫順序優化。而在有Read Request進來的時候分為兩種情況，第一種是內存中完成數據交換。
Send流量從平均10M/s增加到了到平均60M/s，而磁盤Read只有不超過50KB/s。PageCache降低磁盤I/O效果非常明顯。
接下來是讀一些收到了一段時間，已經從內存中被換出刷寫到磁盤上的老數據。
其他指標還是老樣子，而磁盤Read已經飚高到40+MB/s。此時全部的數據都已經是走硬盤了(對硬盤的順序讀取OS層會進行Prefill PageCache的優化)。依然沒有任何性能問題。
Tips
Kafka官方并不建議通過Broker端的log.flush.interval.messages和log.flush.interval.ms來強制寫盤，認為數據的可靠性應該通過Replica來保證，而強制Flush數據到磁盤會對整體性能產生影響。
可以通過調整/proc/sys/vm/dirty_background_ratio和/proc/sys/vm/dirty_ratio來調優性能。
臟頁率超過第一個指標會啟動pdflush開始Flush Dirty PageCache。
臟頁率超過第二個指標會阻塞所有的寫操作來進行Flush。
根據不同的業務需求可以適當的降低dirty_background_ratio和提高dirty_ratio。
Partition
Partition是Kafka可以很好的橫向擴展和提供高并發處理以及實現Replication的基礎。
擴展性方面。首先，Kafka允許Partition在集群內的Broker之間任意移動，以此來均衡可能存在的數據傾斜問題。其次，Partition支持自定義的分區算法，例如可以將同一個Key的所有消息都路由到同一個Partition上去。同時Leader也可以在In-Sync的Replica中遷移。由于針對某一個Partition的所有讀寫請求都是只由Leader來處理，所以Kafka會盡量把Leader均勻的分散到集群的各個節點上，以免造成網絡流量過于集中。
并發方面。任意Partition在某一個時刻只能被一個Consumer Group內的一個Consumer消費(反過來一個Consumer則可以同時消費多個Partition)，Kafka非常簡潔的Offset機制最小化了Broker和Consumer之間的交互，這使Kafka并不會像同類其他消息隊列一樣，隨著下游Consumer數目的增加而成比例的降低性能。此外，如果多個Consumer恰巧都是消費時間序上很相近的數據，可以達到很高的PageCache命中率，因而Kafka可以非常高效的支持高并發讀操作，實踐中基本可以達到單機網卡上限。
不過，Partition的數量并不是越多越好，Partition的數量越多，平均到每一個Broker上的數量也就越多。考慮到Broker宕機(Network Failure, Full GC)的情況下，需要由Controller來為所有宕機的Broker上的所有Partition重新選舉Leader，假設每個Partition的選舉消耗10ms，如果Broker上有500個Partition，那么在進行選舉的5s的時間里，對上述Partition的讀寫操作都會觸發LeaderNotAvailableException。
再進一步，如果掛掉的Broker是整個集群的Controller，那么首先要進行的是重新任命一個Broker作為Controller。新任命的Controller要從Zookeeper上獲取所有Partition的Meta信息，獲取每個信息大概3-5ms，那么如果有10000個Partition這個時間就會達到30s-50s。而且不要忘記這只是重新啟動一個Controller花費的時間，在這基礎上還要再加上前面說的選舉Leader的時間 -_-!!!!!!
此外，在Broker端，對Producer和Consumer都使用了Buffer機制。其中Buffer的大小是統一配置的，數量則與Partition個數相同。如果Partition個數過多，會導致Producer和Consumer的Buffer內存占用過大。
Tips
Partition的數量盡量提前預分配，雖然可以在后期動態增加Partition，但是會冒著可能破壞Message Key和Partition之間對應關系的風險。
Replica的數量不要過多，如果條件允許盡量把Replica集合內的Partition分別調整到不同的Rack。
盡一切努力保證每次停Broker時都可以Clean Shutdown，否則問題就不僅僅是恢復服務所需時間長，還可能出現數據損壞或其他很詭異的問題。
Producer
Kafka的研發團隊表示在0.8版本里用Java重寫了整個Producer，據說性能有了很大提升。我還沒有親自對比試用過，這里就不做數據對比了。本文結尾的擴展閱讀里提到了一套我認為比較好的對照組，有興趣的同學可以嘗試一下。
其實在Producer端的優化大部分消息系統采取的方式都比較單一，無非也就化零為整、同步變異步這么幾種。
Kafka系統默認支持MessageSet，把多條Message自動地打成一個Group后發送出去，均攤后拉低了每次通信的RTT。而且在組織MessageSet的同時，還可以把數據重新排序，從爆發流式的隨機寫入優化成較為平穩的線性寫入。
此外，還要著重介紹的一點是，Producer支持End-to-End的壓縮。數據在本地壓縮后放到網絡上傳輸，在Broker一般不解壓(除非指定要Deep-Iteration)，直至消息被Consume之后在客戶端解壓。
當然用戶也可以選擇自己在應用層上做壓縮和解壓的工作(畢竟Kafka目前支持的壓縮算法有限，只有GZIP和Snappy)，不過這樣做反而會意外的降低效率！！！！Kafka的End-to-End壓縮與MessageSet配合在一起工作效果最佳，上面的做法直接割裂了兩者間聯系。至于道理其實很簡單，壓縮算法中一條基本的原理“重復的數據量越多，壓縮比越高”。無關于消息體的內容，無關于消息體的數量，大多數情況下輸入數據量大一些會取得更好的壓縮比。
不過Kafka采用MessageSet也導致在可用性上一定程度的妥協。每次發送數據時，Producer都是send()之后就認為已經發送出去了，但其實大多數情況下消息還在內存的MessageSet當中，尚未發送到網絡，這時候如果Producer掛掉，那就會出現丟數據的情況。
為了解決這個問題，Kafka在0.8版本的設計借鑒了網絡當中的ack機制。如果對性能要求較高，又能在一定程度上允許Message的丟失，那就可以設置request.required.acks=0 來關閉ack，以全速發送。如果需要對發送的消息進行確認，就需要設置request.required.acks為1或-1，那么1和-1又有什么區別呢？這里又要提到前面聊的有關Replica數量問題。如果配置為1，表示消息只需要被Leader接收并確認即可，其他的Replica可以進行異步拉取無需立即進行確認，在保證可靠性的同時又不會把效率拉得很低。如果設置為-1，表示消息要Commit到該Partition的ISR集合中的所有Replica后，才可以返回ack，消息的發送會更安全，而整個過程的延遲會隨著Replica的數量正比增長，這里就需要根據不同的需求做相應的優化。
Tips
Producer的線程不要配置過多，尤其是在Mirror或者Migration中使用的時候，會加劇目標集群Partition消息亂序的情況(如果你的應用場景對消息順序很敏感的話)。
0.8版本的request.required.acks默認是0(同0.7)。
Consumer
Consumer端的設計大體上還算是比較常規的。
? 通過Consumer Group，可以支持生產者消費者和隊列訪問兩種模式。
? Consumer API分為High level和Low level兩種。前一種重度依賴Zookeeper，所以性能差一些且不自由，但是超省心。第二種不依賴Zookeeper服務，無論從自由度和性能上都有更好的表現，但是所有的異常(Leader遷移、Offset越界、Broker宕機等)和Offset的維護都需要自行處理。
? 大家可以關注下不日發布的0.9 Release。開發人員又用Java重寫了一套Consumer。把兩套API合并在一起，同時去掉了對Zookeeper的依賴。據說性能有大幅度提升哦~~
Tips
強烈推薦使用Low level API，雖然繁瑣一些，但是目前只有這個API可以對Error數據進行自定義處理，尤其是處理Broker異常或由于Unclean Shutdown導致的Corrupted Data時，否則無法Skip只能等著“壞消息”在Broker上被Rotate掉，在此期間該Replica將會一直處于不可用狀態。
那么Kafka如何做到能支持能同時發送大量消息的呢？
答案是Kafka通過批量壓縮和發送做到的。
我們知道消息肯定是放在內存中的，大數據場景消息的不斷發送，內存中不斷存在大量的消息，很容易引起GC
頻繁的GC特別是full gc是會造成“stop the world”，也就是其他線程停止工作等待垃圾回收線程執行，繼而進一步影響發送的速度影響吞吐量，那么Kafka是如何做到優化JVM的GC問題的呢？看完本篇文章你會get到。
Kafka的內存池
下面介紹下Kafka客戶端發送的大致過程，如下圖：
Kafka的kafkaProducer對象是線程安全的，每個發送線程在發送消息時候共用一個kafkaProducer對象來調用發送方法，最后發送的數據根據Topic和分區的不同被組裝進某一個RecordBatch中。
發送的數據放入RecordBatch后會被發送線程批量取出組裝成ProduceRequest對象發送給Kafka服務端。
可以看到發送數據線程和取數據線程都要跟內存中的RecordBatch打交道，RecordBatch是存儲數據的對象，那么RecordBatch是怎么分配的呢？
下面我們看下Kafka的緩沖池結構，如下圖所示：
名詞解釋：緩沖池：BufferPool（緩沖池）對象，整個KafkaProducer實例中只有一個BufferPool對象。內存池總大小，它是已使用空間和可使用空間的總和，用totalMemory表示(由buffer.memory配置，默認32M)。
可使用的空間：它包含包括兩個部分，綠色部分代表未申請未使用的部分，用availableMemory表示
黃色部分代表已經申請但沒有使用的部分，用一個ByteBuffer雙端隊列(Deque)表示，在BufferPool中這個隊列叫free，隊列中的每個ByteBuffer的大小用poolableSize表示(由batch.size配置，默認16k)，因為每次free申請內存都是以poolableSize為單位申請的，申請poolableSize大小的bytebuffer后用RecordBatch來包裝起來。
已使用空間：代表緩沖池中已經裝了數據的部分。
根據以上介紹，我們可以知道，總的BufferPool大小=已使用空間+可使用空間；free的大小=free.size * poolableSize（poolsize就是單位batch的size）。
數據的分配過程?總的來說是判斷需要存儲的數據的大小是否free里有合適的recordBatch裝得下
如果裝得下則用recordBatch來存儲數據，如果free里沒有空間但是availableMemory+free的大小比需要存儲的數據大（也就是說可使用空間比實際需要申請的空間大），說明可使用空間大小足夠，則會用讓free一直釋放byteBuffer空間直到有空間裝得下要存儲的數據位置，如果需要申請的空間比實際可使用空間大，則內存申請會阻塞直到申請到足夠的內存為止。
整個申請過程如下圖：
數據的釋放過程?總的來說有2個入口，釋放過程如下圖：
再來看段申請空間代碼：
?//判斷需要申請空間大小，如果需要申請空間大小比batchSize小，那么申請大小就是batchsize，如果比batchSize大，那么大小以實際申請大小為準
int size = Math.max(this.batchSize, Records.LOG_OVERHEAD + Record.recordSize(key, value));
log.trace("Allocating a new {} byte message buffer for topic {} partition {}", size, tp.topic(), tp.partition());
//這個過程可以參考圖3
ByteBuffer buffer = free.allocate(size, maxTimeToBlock);
再來段回收的核心代碼：
public void deallocate(ByteBuffer buffer, int size) { lock.lock(); try { //只有標準規格（bytebuffer空間大小和poolableSize大小一致的才放入free） if (size == this.poolableSize && size == buffer.capacity()) { //注意這里的buffer是直接reset了，重新reset后可以重復利用，沒有gc問題 buffer.clear(); //添加進free循環利用 this.free.add(buffer); } else { //規格不是poolableSize大小的那么沒有進行重制，但是會把availableMemory增加，代表整個可用內存空間增加了，這個時候buffer的回收依賴jvm的gc this.availableMemory += size; } //喚醒排在前面的等待線程 Condition moreMem = this.waiters.peekFirst(); if (moreMem != null) moreMem.signal(); } finally { lock.unlock(); }}通過申請和釋放過程流程圖以及釋放空間代碼，我們可以得到一個結論
就是如果用戶申請的數據（發送的消息）大小都是在poolableSize（由batch.size配置，默認16k）以內，并且申請時候free里有空間，那么用戶申請的空間是可以循環利用的空間，可以減少gc，但是其他情況也可能存在直接用堆內存申請空間的情況，存在gc的情況。
如何盡量避免呢，如果批量消息里面單個消息都是超過16k，可以考慮調整batchSize大小。
如果沒有使用緩沖池，那么用戶發送的模型是下圖5，由于GC特別是Full GC的存在，如果大量發送，就可能會發生頻繁的垃圾回收，導致的工作線程的停頓，會對整個發送性能，吞吐量延遲等都有影響。
使用緩沖池后，整個使用過程可以縮略為下圖：
總結
Kafka通過使用內存緩沖池的設計，讓整個發送過程中的存儲空間循環利用，有效減少JVM GC造成的影響，從而提高發送性能，提升吞吐量。
與50位技術專家面對面20年技術見證，附贈技術全景圖總結
以上是生活随笔為你收集整理的kafka架构、亿级数据如何优化GC的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： kafka压力测试说明
- 下一篇： 续说零拷贝(Zero-Copy) - D