訊息佇列之Kafka——從架構技術重新理解Kafka
Apache Kafka® 是 一個分散式流處理平臺. 這到底意味著什麼呢?
我們知道流處理平臺有以下三種特性:
- 可以讓你釋出和訂閱流式的記錄。這一方面與訊息佇列或者企業訊息系統類似。
- 可以儲存流式的記錄，並且有較好的容錯性。
- 可以在流式記錄產生時就進行處理。
Kafka適合什麼樣的場景?
它可以用於兩大類別的應用:
- 構造實時流資料管道，它可以在系統或應用之間可靠地獲取資料。 (相當於message queue)
- 構建實時流式應用程式，對這些流資料進行轉換或者影響。 (就是流處理，通過kafka stream topic和topic之間內部進行變化
Kafka有四個核心的API:
- The Producer API 允許一個應用程式釋出一串流式的資料到一個或者多個Kafka topic。
- The Consumer API 允許一個應用程式訂閱一個或多個 topic ，並且對釋出給他們的流式資料進行處理。
- The Streams API 允許一個應用程式作為一個流處理器，消費一個或者多個topic產生的輸入流，然後生產一個輸出流到一個或多個topic中去，在輸入輸出流中進行有效的轉換。
- The Connector API 允許構建並執行可重用的生產者或者消費者，將Kafka topics連線到已存在的應用程式或者資料系統。比如，連線到一個關係型資料庫，捕捉表（table）的所有變更內容。
在Kafka中，客戶端和伺服器使用一個簡單、高效能、支援多語言的 TCP 協議.此協議版本化並且向下相容老版本， 我們為Kafka提供了Java客戶端，也支援許多其他語言的客戶端。
————————————————————————————————————————————————
以上摘自Apache Kafka官網
而本文關注的焦點是：構造實時流資料管道，即message queue部分。也就是我們常使用的“訊息佇列”部分，這部分本身也是Kafka最初及最基本的底層設計。
讓我們回到最初Kafka還沒有設計出來的時候，通過重新設計Kafka，一步步瞭解為什麼Kafka是我們現在看到的樣子，到時我們將瞭解到Kafka作為訊息佇列會高吞吐量、分散式、高容錯穩定。我們把這個專案命名為：Kafka-R。
現在我們開始設計Kafka-R，我們正式設計Kafka-R之前需要考慮設計目標，也就是我的Kafka-R設計出來到底是用來幹嘛的，適用於什麼業務場景，解決什麼需求痛點。
可以很快想到：資料交換。這是訊息佇列的基本功能與要求。
然後呢？可以作為個大平臺，支援多語言，最好能滿足大公司的業務需求，而且最好是實時的，至少是低延遲。
概括起來就是：我們設計Kafka-R的目標是可以作為一個統一的平臺來處理大公司可能擁有的所有實時資料饋送。
為了滿足我們的Kafka-R的設計目標，那麼Kafka-R需要具備以下這些特徵：
具有高吞吐量來支援高容量事件流。
能夠正常處理大量的資料積壓，以便支援來自離線系統的週期性資料載入。
系統必須處理低延遲分發，來處理更傳統的訊息傳遞用例。
資料饋送分割槽與分散式，以及實時。
系統在出現機器故障時能夠保證容錯。
一、資料的儲存方式——in-memory&in-disk
有兩種選擇：第一種，使用in-memory cache，並在空間不足的的時候將資料flush到檔案系統中。
另外一種，使用in-disk，一開始把所有的資料寫入檔案系統的持久化日誌中。
我們的Kafka-R採用in-disk。實際上在此情況資料被轉移到了核心的pagecache中。
“磁碟速度慢”是人們的普遍印象，那麼Kafka-R的資料儲存和快取基於檔案系統，這樣的效能能夠接受嗎？
而事實是，磁碟的速度比人們預期的要慢得多，也快得多，取決於人們使用磁碟的方式。
我們知道磁碟有順序讀和隨機讀兩種模式，之間的效能差異很大，但具體差距多少呢？
使用6個7200rpm、SATA介面、RAID-5的磁碟陣列在JBOD配置下的順序寫入的效能約為600MB/秒，但隨機寫入的效能僅約為100k/秒，相差6000倍。
線性的讀取和寫入是磁碟使用模式中最有規律的，並且作業系統進行了大量的優化。現代作業系統提供了read-ahead和write-behind技術，read-ahead是以大的data block為單位預先讀取資料，而write-hehind將多個小型的邏輯寫合併成一次大型的物理磁碟寫入。
磁碟除了訪問模式，還有兩個低效率操作影響系統的效能：大量的小型I/O操作，過多的位元組拷貝。
那麼我們怎麼處理這些問題呢？
針對於大量的小型I/O操作，Kafka-R使用“訊息塊”將訊息合理分組。使網路請求將多個訊息打包成一組，而不是每次傳送一條訊息，從而使整組訊息分擔網路往返的開銷。
另一個過多的位元組拷貝，Kafka-R使用producer，broker和consumer都共享的標準化通用的二進位制訊息格式，這樣資料塊不用修改就能在他們之間傳遞。
保持這種通用的格式有什麼用呢？
可以對持久化日誌塊的網路傳輸進行優化。現代的unix作業系統提供了一個高度優化的編碼方式，用於將資料從pagecache轉移到socket網路連線中。
資料從檔案到套接字的常見資料傳輸過程：磁碟->pagecache->使用者空間快取區->套接字緩衝區（核心空間）->NIC快取區
1. 作業系統從磁碟讀區資料到核心空間的pagecache
2. 應用程式讀取核心空間的資料到使用者空間的快取區
3. 應用程式將資料（使用者空間的快取區）寫會核心空間到套接字緩衝區（核心空間）
4. 作業系統將資料從套接字緩衝區（核心空間）複製到能夠通過網路傳送的NIC緩衝區
共進行了4次copy操作和2次系統呼叫，顯然很低效。在Linux系統中使用zero-copy（零拷貝）優化，其中之一sendfile，使用後的資料傳輸過程是這樣：磁碟->pagecache->NIC快取區。
我們的Kafka-R通過使用zero-copy優化技術，可以用盡可能低的消費代價讓多個consumer消費。資料在使用時只會被複制到pagecache中一次，這樣訊息能夠以接近網路連線的速度上限進行消費。
二、資料結構——BTree&日誌解決方案
日誌解決方案即簡單讀取與追加來操作檔案。
我們的Kafka-R採用日誌解決方案。
我們知道BTree是通用的資料結構，其廣泛用於隨機的資料訪問。BTree的操作時間複雜度是O(log N)，基本等同於常數時間，但在磁碟上則不成立。
每個磁碟同時只能執行一次定址，並行性受到限制。少量的磁碟定址也有很高的開銷。資料翻倍時效能下降不止兩倍。
而日誌解決方案的資料儲存架構，所有的操作時間複雜度都是O(1)，並且讀不會阻塞寫，讀之間也不會相互影響。
由於效能和資料的大小是完全分離的，則伺服器可以使用大量廉價、低轉速的1+TB SATA硬碟，即使這些硬碟的定址效能很差，在大規模讀寫的效能也可以接受，而且三分之一的價格三倍的容量。
三、獲取資料方式——push-based&pull-based
由consumer從broker那裡pull資料呢？還是從broker將資料push到consumer?
我們的Kafka-R採用pull-based方式。
這是大多數訊息系統所共享的傳統的方式：即producer把資料push到broker，然後consumer從broker中pull資料。
push-based系統優點：
1. 讓consumer能夠以最大速率消費。
push-based系統缺點：
1. 由於broker控制著資料傳輸速率，所以很難處理不同的consumer。
2. 當消費速率低於生產速率時，consumer往往會不堪重負（本質類似於拒絕服務攻擊）。
3. 必須選擇立即傳送請求或者積累更多的資料，然後在不知道下游的consumer能否立即處理它的情況下發送這些資料。特別系統為低延遲狀態下，這樣會極度糟糕浪費。
pull-based系統優點：
1. 可以大批量生產要傳送給consumer的資料。
pull-based系統缺點：
1. 如果broker中沒有資料，consumer可能會在一個緊密的迴圈中結束輪詢，實際上會busy-waiting直到資料到來。
為了避免busy-waiting，我們的Kafka-R的pull引數重加入引數，使得consumer在一個“long pull”中阻塞等待，知道資料到來（還可以選擇等待給定位元組長度的資料來確保傳輸長度）。
四、消費者的位置——consumed&offset
Kafka-R的消費過程：consumer通過向broker發出一個“fetch”請求來獲取它想要消費的partition。consumer的每個請求在log中指定了對應的offset，並接收從該位置開始的一大塊資料。
consumed指通過狀態標示已經被消費的資料。
大多數訊息系統都在broker上儲存被消費訊息的元資料。當訊息被傳遞給consumer，broker要麼立即在本地記錄該事件，要麼等待consumer的確認後再記錄。
消費者的位置問題其實就是broker和consumer之間被消費資料的一致性問題。如果broker再每條訊息被髮送到網路的時候，立即將其標記為consumd，那麼一旦consumer無法處理該訊息（可能由consumer崩潰或者請求超時或者其他原因導致），該訊息就會丟失。為了解決訊息丟失的問題，許多訊息系統增加了確認機制：即當訊息被髮送出去的時候，訊息被標記為sent而不是consumed；然後broker會等待一個來自consumer的特定確認，再將訊息標記為consumed。這個策略修復了訊息丟失的問題，但也產生了新問題。首先，如果consumer處理了訊息但在傳送確認之前出錯了，那麼該訊息就會被消費兩次。第二個是有關效能的，broker必須為每條訊息儲存多個狀態（首先對其加鎖，確保該訊息只被傳送一次，然後將其永久的標記為consumed，以便將其移除）。還有更棘手的問題，比如如何處理已經發送但一直等不到確認的訊息。
Kafka-R使用offse來處理訊息丟失問題。topic被分割成一組完全有序的partition，其中每一個partition在任意給定的時間內只能被每個訂閱了這個topic的consumer組中的一個consumer消費。意味著partition中每一個consumer的位置僅僅是一個數字，即下一條要消費的訊息的offset。這樣就可以按非常低的代價實現和訊息確認機制等同的效果。consumer還可以回退到之前的offset再次消費之前的資料，這樣的操作違背了佇列的基本原則，但事實證明對consumer來說是個很重要的特性。如果consumer程式碼由bug，並且在bug被發現之前有部分資料被消費了，consumer可以在bug修復後通過回退到之前的offset再次消費這些資料。
五、leader選舉——多數投票機制f+1&ISR
Kafka-R動態維護了一個同步狀態的備份的集合（a set of in-sync replicas），簡稱ISR。
在瞭解ISR之前我們需要先了解in-sync。
Kafka-R判斷節點是否存活有兩種方式：
1. 節點必須可以維護和ZooKeeper的連線，ZooKeeper通過心跳機制檢查每個節點的連線。
2. 如果節點是個follower，它必須能及時的同步leader的寫操作，並且延時不能太久。
只有滿足上面兩個條件的節點就處於“in sync”狀態。leader會追蹤所有“in sync”的節點，如果有節點掛掉了，或是寫超時，或是心跳超時，leader就會把它從同步副本列表中移除。
在ISR集合中節點會和leader保持高度一致，只有這個集合的成員才有資格被選舉為leader，一條訊息必須被這個集合所有節點讀取並追加到日誌中了，這條訊息才能視為提交。
ISR集合發生變化會在ZooKeeper持久化，所以這個集合中的任何一個節點都有資格被選為leader。
多數投票機制f+1顧名思義：假設我們有2f+1個副本，如果在leader宣佈訊息提交之前必須有f+1個副本收到該訊息，並且如果我們從這隻少f+1個副本之中，有著最完整的日誌記錄的follower裡來選擇一個新的leader，那麼在故障數小於f的情況下，選舉出的leader保證具有所有提交的訊息。
多數投票演算法必須處理許多細節，比如精確定義怎樣使日誌更加完整，確保在leader down期間，保證日誌一致性或者副本伺服器的副本集改變。
多數投票機制有一個非常好的優點：延遲取決於較快的伺服器。也就是說，如果副本數是3，則備份完成的等待時間取決於最快的follwer。
因此提交時能避免最慢的伺服器，這也是多數投票機制的優點。
同樣多數投票的缺點也很明顯，多數的節點掛掉後不能選擇出leader。而通過冗餘來避免故障率，會降低吞吐量，不利於處理海量資料。
是一種Quorum讀寫機制（如果選擇寫入時候需要保證一定數量的副本寫入成功，讀取時需要保證讀取一定數量的副本，讀取和寫入之間有重疊）。
Kafka-R保證只要有只少一個同步中的節點存活，提交的訊息就不會丟失。
在一次故障生存之後，大多數的quorum需要三個備份節點和一次確認，ISR只需要兩個備份節點和一次確認。
建立副本的單位是topic的partition,正常情況下，每個分割槽都有一個leader和零或多個follower。總的副本數是包括leader與所有follwer的總和。所有的讀寫操作都由leader處理，一般partition的數量都比broker的數量多的多，各分割槽的leader均勻分佈在broker中。所有的follower節點都同步leader節點的日誌，日誌中的訊息和偏移量都和leader保持一致。
六、Uclean leader選舉——ISR副本&第一個副本
如果節點全掛了的服務恢復。
Kafka-R對於資料不會丟失時基於只少一個節點保持同步狀態，而一旦分割槽上的所有備份節點都掛了，就無法保證了。
Kafka-R預設“第一個副本”策略。
ISR副本：等待一個ISR的副本重新恢復正常服務，並選擇這個副本作為新leader（極大可能擁有全部資料）
第一個副本：選擇第一個重新恢復正常服務的副本（不一定是ISR）作為leader。
這是可用性和一致性之間的簡單妥協，如果只等待ISR的備份節點，只要ISR備份節點都掛了，那麼服務都一直會不可用，如果他們的資料損壞了或者丟失了，那就會是長久的宕機。另一方面，如果不是ISR中的節點恢復服務並且我們允許它成為leader，那麼它的資料就是可信的來源，即使它不能保證記錄了每一個已經提交的訊息。
可以配置屬性unclean.leader.election.enable禁用次策略，那麼就會使用“ISR副本”策略即停機時間優於不同步，以修改預設配置。
通過以上的架構技術的分析和選型，我們就大致設計出了我們的訊息佇列Kafka-R。