Apache Kafka之設計

轉自: http://blog.csdn.net/kevin_hx001/article/details/9413565

       http://kafka.apache.org/design.html

咱們爲何要構建這個系統

Kafka是一個分佈式、分區的、多副本的、多訂閱者的「提交」日誌系統。

咱們構建這個系統是由於咱們認爲，一個實現無缺的操做日誌系統是一個最基本的基礎設施，它能夠替代一些系統來做諸如：消息處理，ETL（Extraction-Transformation-Loading），日誌收集，流式處理等工做。咱們的目標就是能有一個擁有足夠吞吐量和能力的系統來將上面這些事情統一在一個平臺上。

活動流數據是任何網站的一部分，這部分數據用來彙報站點的應用狀況。這些數據包括：PV，哪些信息被展現給用戶，搜索詞等。這些信息一般是這樣處理的：將它們以日誌的形式存儲到一些文件中，而後按期地對這些文件進行分析。系統運行數據包括服務器的運行狀況（CPU，IO，請求時間，服務日誌等等），收集這些數據也有許多不一樣的方法。

近年來，活動與運行數據已經成爲站點的關鍵部分，稍微複雜一點的基礎設施也就有了產生的需求。

活動流數據和運行數據的應用場景

1）News feed」 features，將活動廣播給你的朋友

2）經過評分，投票，點擊來肯定哪些項目的集合是有關聯的。

3）安全方面：站點須要阻止無良的爬蟲，限制性api，檢測惡意訪問以及其餘一些檢測和預防系統。

4）運行監測：大多數站點須要一些實時的、可靠的監控來跟蹤運行狀況，以便發生故障的時候觸發報警器。

5）報表與批處理：將數據導入到數據倉庫或者hadoop系統中，從而進行離線分析和報表生成以便商業決策。

活動流數據的特色

傳統的日誌文件收集對於離線的應用場景好比報表生成和批處理都有很好的支持，可是對於實時的處理有很高的延時和很高的計算複雜度。另一方面，現存的消息和隊列系統對實時與近實時的應用場景都是ok的，可是不能很好地處理大量的未消費隊列，持久化經常是過後纔想到。

當向離線系統好比hadoop這樣的系統發數據時就會產生問題，這些系統只會每隔一小時或一天才會去一些數據源拉數據。Kafka的目的就是構建一個隊列平臺可以支持離線與在線的應用場景。

Kafka支持比較通用的消息語義。沒有什麼被綁定到活動處理上，儘管那是咱們的motivating 應用場景。

部署

下圖簡單地展現了在LinkedIn內部的部署拓撲。

須要注意的是一個kafka集羣處理來自不一樣數據源的活動數據。這就爲離線和在線消費者（consumer）提供了一個單一的數據流水線。這一層爲在線活動和異步處理提供了一層緩存。咱們還用kafka來將數據複製到不一樣的數據倉庫，以便離線處理。

咱們不想讓一個kafka集羣跨越全部的數據中心，可是kafka是支持多數據中心的數據流拓撲結構。這能夠經過在集羣之間「鏡像」或者「同步」來實現。這個特性很是簡單，只要將鏡像集羣做爲源集羣的消費者。這就意味着能夠將多個數據中心的數據集中到一個集羣中來。下面是一個例子。

注意到在這兩個集羣裏，各個節點之間是沒有對應關係的。兩個集羣的大小有可能不同，包含的節點數也不同。一個節點能夠鏡像任意數目的源集羣。

主要設計元素

有一系列的設計決策使得kafka與其餘的消息系統不同：

1）將消息持久化做爲一種常見case

2）吞吐量是首要設計約束

3）消費狀態被保存在消費者上而不是服務器上

4）分佈式。生產者，broker，消費者均可以分佈在不一樣的機器上。

這裏的每個特性在下面會詳細地講到。

基本要素

首先是一些基本的術語和概念。

消息是通信的基本單元。消息被生產者發佈到一個主題，也就是說被物理上發佈到一個叫broker的服務器上。必定數量的消費者註冊到一個主題，每一個發佈到這個主題的消息會遞送給這些消費者。

kafka是分佈式的——生產者，消費者，brokers均可以跑在一個集羣上做爲一個邏輯上的組而協做着。這對broker和生產者來講是至關天然的，但對消費者來講還須要額外的一些支持。每一個消費者進程屬於一個消費者組，每條消息只會傳遞給組內的一個進程。所以一個消費者組容許多個進程或者機器做爲邏輯上的一個消費者運行。消費者組的概念是至關牛逼的，它能夠用來支持隊列語義或者JMS中的主題語義。若是是隊列語義，咱們能夠將全部的消費者放到一個消費者組中，這種狀況下，每條消息只會到達一個消費者。在主題語義中，每一個消費者自成一組，這樣，全部的消費者會接收到每一條消息。在咱們的應用中，更通常的狀況是，咱們有多個邏輯上的組，每一個組由多臺機器組成，它們邏輯上做爲一個總體運行。在大數據狀況下，Kafka還有一個更好的特性：無論一個主題有多少個消費者，一條消息只會被存一次。

消息持久化和緩存

不要害怕文件系統

Kafka高度依賴文件系統來存儲和緩存消息。通常的人都認爲「磁盤是緩慢的」，這使得人們對「持久化結構提供具備競爭性的性能」這樣的結論持有懷疑態度。實際上，磁盤比人們預想的快不少也慢不少，這取決於它們如何被使用；一個好的磁盤結構設計可使之跟網絡速度同樣快。

一個有關磁盤性能的關鍵事實是：磁盤驅動器的吞吐量跟尋道延遲是相背離的。結果就是：在一個6 7200rpm SATA RAID-5 的磁盤陣列上線性寫的速度大概是300M/秒，可是隨機寫的速度只有50K/秒，二者相差將近10000倍。線性讀寫在大多數應用場景下是能夠預測的，所以，操做系統利用read-ahead和write-behind技術來從大的數據塊中預取數據，或者將多個邏輯上的寫操做組合成一個大寫物理寫操做中。更多的討論能夠在ACM Queue Artical中找到，他們發現，對磁盤的線性讀在有些狀況下能夠比內存的隨機訪問要快一些。

爲了補償這個性能上的分歧，現代操做系統在內存和磁盤緩存的利用上變得很是aggressive。如今操做系統會很是開心地將全部空閒的內存做爲磁盤緩存，儘管在內存回收的時候會有一點性能上的代價。全部的磁盤讀寫操做會在這個統一的緩存上進行。這個特性不太空易被關掉，除非用直接IO的方法，因此儘管一個進程維護着一個進程內的數據緩。存，這些數據仍是會在OS的頁緩存中被複制，實際上就是全部的數據都保存了兩次。

此外，咱們是在JVM的基礎上構建的，熟悉java內存應用管理的人應該清楚如下兩件事情：

1）一個對象的內存消耗是很是高的，常常是所存數據的兩倍或者更多。

2）隨着堆內數據的增多，Java的垃圾回收會變得很是昂貴。

基於這些事實，利用文件系統而且依靠頁緩存比維護一個內存緩存或者其餘結構要好——咱們至少要使得可用的緩存加倍，經過自動訪問可用內存，而且經過存儲更緊湊的字節結構而不是一個對象，這將有可能再次加倍。這麼作的結果就是在一臺32GB的機器上，若是不考慮GC懲罰，將最多有28-30GB的緩存。此外，這些緩存將會一直存在即便服務重啓，然而進程內緩存須要在內存中重構（10GB緩存須要花費10分鐘）或者它須要一個徹底冷緩存啓動（很是差的初始化性能）。它同時也簡化了代碼，由於如今全部的維護緩存和文件系統之間內聚的邏輯都在操做系統內部了，這使得這樣作比one-off in-process attempts更加高效與準確。若是你的磁盤應用更加傾向於順序讀取，那麼read-ahead在每次磁盤讀取中實際上獲取到這人緩存中的有用數據。

以上這些建議了一個簡單的設計：不一樣於維護儘量多的內存緩存而且在須要的時候刷新到文件系統中，咱們換一種思路。全部的數據不須要調用刷新程序，而是馬上將它寫到一個持久化的日誌中。事實上，這僅僅意味着，數據將被傳輸到內核頁緩存中並稍後被刷新。咱們能夠增長一個配置項以讓系統的用戶來控制數據在何時被刷新到物理硬盤上。

常數時間就知足要求

消息系統元數據的持久化數據結果常常是一個B樹。B樹是一個很好的結構，能夠用在事務型與非事務型的語義中。可是它須要一個很高的花費。B樹的操做須要O(logN)。一般狀況下，這被認爲與常數時間等價，但這對磁盤操做來講是不對的。磁盤尋道一次須要10ms，而且一次只能尋一個，所以並行化是受限的。

直覺上來說，一個持久化的隊列能夠構建在對一個文件的讀和追加上，就像通常狀況下的日誌解決方案。儘管和B樹相比，這種結構不能支持豐富的語義，可是它有一個優勢，全部的操做都是常數時間，讀數據不會阻塞寫數據。

事實上幾乎無限制的磁盤訪問意味着咱們能夠提供通常消息系統沒法提供的特性。好比說，消息被消費後不是立馬被刪除，咱們能夠將這些消息保留一段相對比較長的時間（好比一個星期）。

效率最大化

咱們的假設是，消息的數量是至關大的，事實上是這個站點的一些page views。此外，咱們假設，每一條被髮布的消息至少被讀一次（常常是屢次），所以咱們只去優化消費而不是生產。

通常狀況下有兩種狀況會致使低效：大多的網絡請求，過多的字節拷貝。

爲了提升效率，API的構建是圍繞消息集合的。一次網絡請求發一個消息集合，而不是每一次只發一條消息。

MessageSet的實現自己是一個很是簡單的API，它將一個字節數組或者文件進行打包。因此對消息的處理，這裏沒有分開的序列化和反序列化的上步驟，消息的字段能夠按需反序列化（若是沒有須要，能夠不用反序列化）。

由broker保存的消息日誌自己只是一個消息集合的目錄，這些消息已經被寫入磁盤。這種抽象容許單一一個字節能夠被broker和消費者所分享（某種程度上生產者也能夠，儘管生產者那頭的消息只有再被計算過校驗和以後纔會加入到日誌中去）。

維護這樣的通用格式對能夠對大多數重要的操做進行優化：持久日誌數據塊的網絡傳輸。如今的Unix操做系統提供一種高優化的代碼路徑將數據從頁緩存傳到一個套接字（socket）；在Linux中，這能夠經過調用sendfile系統調用來完成。Java提供了訪問這個系統調用的方法：FileChannel.transferTo api。

爲了理解sendfile的影響，須要理解通常的將數據從文件傳到套接字的路徑：

1）操做系統將數據從磁盤讀到內核空間的頁緩存中

2）應用將數據從內核空間讀到用戶空間的緩存中

3）應用將數據寫回內存空間的套接字緩存中

4）操做系統將數據從套接字緩存寫到網卡緩存中，以便將數據經網絡發出

這樣作明顯是低效的，這裏有四次拷貝，兩次系統調用。若是使用sendfile，再次拷貝能夠被避免：容許操做系統將數據直接從頁緩存發送到網絡上。因此在這個優化的路徑中，只有最後一步將數據拷貝到網卡緩存中是須要的。

咱們指望一個主題上有多個消費者是一種常見的應用場景。利用上述的零拷貝，數據只被拷貝到頁緩存一次，而後就能夠在每次消費時被重得利用，而不須要將數據存在內存中，而後在每次讀的時候拷貝到內核空間中。這使得消息消費速度能夠達到網絡鏈接的速度。

端到端的批量壓縮

在許多場景下，瓶頸實際上不是CPU而是網絡。這在須要在多個數據中心之間發送消息的數據流水線的狀況下更是如此。固然，用戶能夠不須要Kafka的支持而發送壓縮後的消息，可是這會致使很是差的壓縮率。高效的壓縮須要將多個消息一起壓縮而不是對每個消息進行壓縮。理想狀況下，這能夠在端到端的狀況下實現，數據會先被壓縮，而後被生產者發送，而且在服務端也是保持壓縮狀態，只有在最終的消費者端纔會被解壓縮。

Kafka經過遞歸消息集合來支持這一點。一批消息能夠放在一塊兒被壓縮，而後以這種形式發給服務器。這批消息會被遞送到相同的消費者那裏，而且保持壓縮的形式，直到它到達目的地。

Kafka支持GZIP和Snappy壓縮協議，更多的細節能夠在這裏找到：https://cwiki.apache.org/confluence/display/KAFKA/Compression

消費者狀態

在Kafka中，消費者負責記錄狀態信息（偏移量），也就是已經消費到哪一個位置了。準確地說，消費者庫將他們的狀態信息寫到zookeeper中。可是，將狀態數據寫到另外一個地方——處理結果所存放的數據中心——可能會更好。打個比方，消費者可能只須要簡單地將一些合計值寫到中心化的事務型OLTP數據庫中。在這種狀況下，消費者能夠將狀態信息寫到同一個事務中。這解決了分佈式一致性問題——經過去除分佈式部分。相似的技巧能夠用在一些非事務型的系統中。一個搜索系統能夠將消費者狀態存放在索引塊中。儘管這不提供持久性保證，但這意味着索引能夠和消費者狀態保持同步：若是一個沒有刷新的索引塊在一次故障中丟失了，那麼這些索引能夠從最近的檢查點偏移處開始從新消費。一樣的，在並行加載數據到Hadoop時，能夠利用相似的技巧。每一個mapper在map 任務的最後將偏移量寫到HDFS中。這樣的話，若是一個加載任務失敗了，每一個mapper能夠簡單地從存儲在HDFS中的偏移量處重啓消費。

這個決定有另一個好處。消費者能夠從新消費已經消費過的數據。這違反了隊列的性質，可是這樣可使多個消費者一塊兒來消費。打個比方，若是一段消費者代碼出bug了，在發現bug之間這個消費者又消費了一堆數據，那個在bug修復以後，消費者能夠從指定的位置從新消費。

拉仍是推?

Kafka採用的策略是：生產者把數據推到borker上，而消費者主動去broker上拉數據。最近的一些系統包括flume和scribe，都是broker將數據推給消費者，這有可能會存在一個問題，若是推的速度過快，消費者會被淹沒。而在Kafka中不會出現這樣的問題，由於消費者是主動去borker上拉數據的。

分佈式

沒有一箇中心節點，broker之間是對等的，broker能夠隨時添加與刪除。相似的，生產者與消費者能夠在任什麼時候間動態啓動。每一個borker在zookeeper上註冊一些元數據。生產者與消費者能夠利用zookeeper來發現主題，而且在生產與消費之間協調。關於這一點的細節會在下面講到。

生產者

自動的生產者負載均衡

Kafka支持消息生產者在客戶端的負載均衡，或者利用專有的負載均衡器來均衡TCP鏈接。一個專用的四層均衡器經過將TCP鏈接均衡到Kafka的broker上來工做。在這種配置下，全部的來自同一個生產者的消息被髮送到一個borker上，這種作法的優勢是，一個生產者只須要一個TCP鏈接，而不須要與zookeeper的鏈接。缺點是負載均衡只能在TCP鏈接的層面上來作，所以，它有可能不是均衡得很是好（若是一些生產者比其餘生產者生產更多的消息，給每一個broker分配相同的TCP鏈接不必定會使每一個broker獲得相同的消息）。

基於zookeeper的客戶端的負載均衡能夠解決這個問題。它容許生產者動態地發現新的broker，而且在每一個請求上進行負載均衡。一樣的，它容許生產者根據一些鍵將數據分開，而不是隨機分，這能夠增長與消費者的粘性（好比，根據用用戶id來化分數據的消費）。這個特性被稱爲「語義化分」，下文會詳述。

這種基於zookeeper的負載均衡以下所述。zookeeper watchers註冊如下一些事件：

1）一個新的broker啓動

2）一個broker關閉

3）一個新的主題註冊進來

4）一個borker註冊一個已經存在的主題

在內部，生產者維護一個與borker的彈性鏈接池。這個鏈接池經過zookeeper watchers的回調函數來保持更新以便與全部存活的broker創建或保持鏈接。當一個生產者對某一個主題的請求上來時，一個主題的分區被分區器提取到。鏈接池中的一個鏈接被用來將數據發送到前面所選的那個broker分區中。

異步發送

異步的非阻塞發送對於擴展消息系統是基本的。在Kafka中，生產者提供一個選項用來使用生產請求的異步分派（producer.type=async）。這容許將生產請求緩存在一個內存隊列中，而後在被一個時間間隔或者預先設定的batch大小觸發時發送出去。因爲數據是從異構的機器上以不一樣的速率發佈的，這種異步的緩存機制能夠生成統一的通往broker的traffic， 從而使得網絡資源獲得充分利用，同時也提升吞吐量。