Kafka基本知識整理

首先Kafka是一個分散式訊息佇列中介軟體，Apache頂級專案，https://kafka.apache.org/   高效能、持久化、多副本備份、橫向擴充套件。

生產者Producer往佇列裡傳送訊息，消費者Consumer從佇列裡消費訊息，然後進行業務邏輯。應用場景主要有：解耦、削峰（緩衝）、非同步處理、排隊、分散式事務控制等等。

1. Kafka對外使用Topic(主題)的概念，生產者往Topic裡寫訊息，消費者從Topic中消費讀訊息。
2. 為了實現水平擴充套件，一個Topic實際是由多個Partition（分割槽）組成的，遇到瓶頸時，可以透過增加Partition的數量來進行橫向擴容。
3. 單個Parition內是保證訊息有序。持久化時，每收到一條訊息，Kafka就是在對應的日誌檔案Append寫，所以效能非常高。

Kafka Data Flow 訊息流轉圖

上圖中，訊息生產者Producers往Brokers裡面的指定Topic中寫訊息，訊息消費者Consumers從Brokers裡面消費指定Topic的訊息，然後進行業務處理。

在實際的部署架構中，Broker、Topic、Partition這些元資料儲存在ZooKeeper中，Kafka的監控、訊息路由（分割槽）由ZooKeeper控制。0.8版本的OffSet也由ZooKeeper控制。

一、訊息生產/傳送過程

Kafka建立Message、傳送時要指定對應的Topic和Value（訊息體），Key（分割槽鍵）和Partition（分割槽）是可選引數。

呼叫Producer的Send()方法後，訊息先進行序列化（訊息序列化器可自定義實現：例如：Protobuf），然後按照Topic和Partition，臨時放到記憶體中指定的傳送佇列中。達到閾值後，然後批次傳送。

傳送時，當Partition沒設定時，如果設定了Key-分割槽鍵（例如：單據型別），按照Key進行Hash取模，保證相同的Key傳送到指定的分割槽Partition。如果未設定分割槽鍵Key，使用Round-Robin輪詢隨機選分割槽Partition。

二、分割槽Partition的高可用和選舉機制

分割槽有副本的概念，保證訊息不丟失。當存在多副本的情況下，會儘量把多個副本，分配到不同的broker上。

Kafka會為Partition選出一個Leader Broker（透過ZooKeeper），之後所有該Partition的請求，實際操作的都是Leader，然後再同步到其他的Follower。

當一個Kafka Broker宕機後，所有Leader在該Broker上的Partition都會重新選舉，在剩餘的Follower中選出一個Leader，繼續提供服務。

正如上面所講：Kafka使用ZooKeeper在多個Broker中選出一個Controller，用於Partition分配和Leader選舉。以下是Partition的分配機制：

• 將所有Broker（假設共n個Broker）和待分配的Partition排序
• 將第i個Partition分配到第（i mod n）個Broker上 （這個就是leader）
• 將第i個Partition的第j個Replica分配到第（(i + j) mode n）個Broker上

Controller會在ZooKeeper的/brokers/ids節點上註冊Watch，一旦有broker宕機，它就能知道。

當Broker宕機後，Controller就會給受到影響的Partition選出新Leader。

Controller從ZooKeeper的/brokers/topics/[topic]/partitions/[partition]/state中，讀取對應Partition的ISR（in-sync replica已同步的副本）串列，選一個出來做Leader。

選出Leader後，更新ZooKeeper的儲存，然後傳送LeaderAndISRRequest給受影響的Broker進行通知。

如果ISR串列是空，那麼會根據配置，隨便選一個replica做Leader，或者乾脆這個partition就是宕機了。

如果ISR串列的有機器，但是也宕機了，那麼還可以等ISR的機器活過來。

多副本同步：

服務端這邊的處理是Follower從Leader批次拉取資料來同步。但是具體的可靠性，是由生產者來決定的。

生產者生產訊息的時候，透過request.required.acks引數來設定資料的可靠性。

 在acks=-1的時候，如果ISR少於min.insync.replicas指定的數目，那麼就會傳回不可用。

 這裡ISR串列中的機器是會變化的，根據配置replica.lag.time.max.ms，多久沒同步，就會從ISR串列中剔除。以前還有根據落後多少條訊息就踢出ISR，在1.0版本後就去掉了，因為這個值很難取，在高峰的時候很容易出現節點不斷的進出ISR串列。

 從ISA中選出leader後，follower會從把自己日誌中上一個高水位後面的記錄去掉，然後去和leader拿新的資料。因為新的leader選出來後，follower上面的資料，可能比新leader多，所以要擷取。這裡高水位的意思，對於partition和leader，就是所有ISR中都有的最新一條記錄。消費者最多隻能讀到高水位；

 從leader的角度來說高水位的更新會延遲一輪，例如寫入了一條新訊息，ISR中的broker都fetch到了，但是ISR中的broker只有在下一輪的fetch中才能告訴leader。

 也正是由於這個高水位延遲一輪，在一些情況下，kafka會出現丟資料和主備資料不一致的情況，0.11開始，使用leader epoch來代替高水位。

三、訊息消費過程

訂閱topic是以一個消費組來訂閱的，一個消費組裡面可以有多個消費者。同一個消費組中的兩個消費者，不會同時消費一個partition。

換句話來說，就是一個partition，只能被消費組裡的一個消費者消費，但是可以同時被多個消費組消費。因此，如果消費組內的消費者如果比partition多的話，那麼就會有個別消費者一直空閑。

 訊息Offset偏移量(訊息的順序號)管理

 一個消費組消費partition，需要儲存offset記錄消費到哪，以前儲存在zk中，由於zk的寫效能不好，以前的解決方法都是consumer每隔一分鐘上報一次。

 ZooKeeper的效能嚴重影響了消費的速度，而且很容易出現重覆消費。

 在0.10版本後，Kafka把這個offset的儲存，從ZooKeeper總剝離，儲存在一個名叫__consumeroffsets topic的Topic中。

 訊息的key由groupid、topic、partition組成，value是偏移量offset。topic配置的清理策略是compact。總是保留最新的key，其餘刪掉。

 一般情況下，每個key的offset都是快取在記憶體中，查詢的時候不用遍歷partition，如果沒有快取，第一次就會遍歷partition建立快取，然後查詢傳回。

 Partitin的Rebalance

 生產過程中broker要分配partition，消費過程這裡，也要分配partition給消費者。類似broker中選了一個controller出來，消費也要從broker中選一個coordinator，用於分配partition。

 coordinator的選舉過程

1. 看offset儲存在那個partition
2. 該partition leader所在的broker就是被選定的coordinator

 互動流程

1. consumer啟動、或者coordinator宕機了，consumer會任意請求一個broker，傳送ConsumerMetadataRequest請求，broker會按照上面說的方法，選出這個consumer對應coordinator的地址。
2. consumer 傳送heartbeat請求給coordinator，傳回IllegalGeneration的話，就說明consumer的資訊是舊的了，需要重新加入進來，進行reblance。傳回成功，那麼consumer就從上次分配的partition中繼續執行。

  Rebalance

1. consumer給coordinator傳送JoinGroupRequest請求。
2. 這時其他consumer發heartbeat請求過來時，coordinator會告訴他們，要reblance了。
3. 其他consumer傳送JoinGroupRequest請求。
4. 所有記錄在冊的consumer都發了JoinGroupRequest請求之後，coordinator就會在這裡consumer中隨便選一個leader。然後回JoinGroupRespone，這會告訴consumer你是follower還是leader，對於leader，還會把follower的資訊帶給它，讓它根據這些資訊去分配partition
5. consumer向coordinator傳送SyncGroupRequest，其中leader的SyncGroupRequest會包含分配的情況。
6. coordinator回包，把分配的情況告訴consumer，包括leader。

   當partition或者消費者的數量發生變化時，都得進行reblance。

   列舉一下會reblance的情況：

1. 增加Partition
2. 增加消費者
3. 消費者主動關閉
4. 消費者宕機
5. coordinator宕機

四、訊息投遞語意

kafka支援3種訊息投遞語意，
At most once：最多一次，訊息可能會丟失，但不會重覆
At least once：最少一次，訊息不會丟失，可能會重覆
Exactly once：只且一次，訊息不丟失不重覆，只且消費一次（0.11中實現，僅限於下游也是kafka）

At least once：（業務中使用比較多）

先獲取資料，再進行業務處理，業務處理成功後commit offset。

1. 生產者生產訊息異常，訊息是否成功寫入不確定，重做，可能寫入重覆的訊息
2. 消費者處理訊息，業務處理成功後，更新offset失敗，消費者重啟的話，會重覆消費

At most once：

先獲取資料，再commit offset，最後進行業務處理。

1. 生產者生產訊息異常，不管，生產下一個訊息，訊息就丟了
2. 消費者處理訊息，先更新offset，再做業務處理，做業務處理失敗，消費者重啟，訊息就丟了。

Exactly once：

首先要保證訊息不丟，再去保證不重覆。所以盯著At least once的原因來搞。

1. 生產者重做導致重覆寫入訊息—-生產保證冪等性
2. 消費者重覆消費—消滅重覆消費，或者業務介面保證冪等性重覆消費也沒問題

業務處理的冪等性非常重要。Kafka控制不了，需要業務來實現。比如所判斷訊息是否已經處理。

解決重覆消費有兩個方法：

1. 下游系統保證冪等性，重覆消費也不會導致多條記錄。
2. 把commit offset和業務處理系結成一個事務。

生產的冪等性：

為每個producer分配一個pid，作為該producer的唯一標識。producer會為每一個維護一個單調遞增的seq。類似的，broker也會為每個記錄下最新的seq。當req_seq == broker_seq+1時，broker才會接受該訊息。因為：

1. 訊息的seq比broker的seq大超過時，說明中間有資料還沒寫入，即亂序了。

2. 訊息的seq不比broker的seq小，那麼說明該訊息已被儲存。

  事務性和原子性

   場景是這樣的：

1. 先從多個源topic中獲取資料。
2. 做業務處理，寫到下游的多個目的topic。
3. 更新多個源topic的offset。

   其中第2、3點作為一個事務，要麼全成功，要麼全失敗。這裡得益與offset實際上是用特殊的topic去儲存，這兩點都歸一為寫多個topic的事務性處理。

   

   引入tid（transaction id），和pid不同，這個id是應用程式提供的，用於標識事務，和producer是誰並沒關係。就是任何producer都可以使用這個tid去做事務，這樣進行到一半就死掉的事務，可以由另一個producer去恢復。
同時為了記錄事務的狀態，類似對offset的處理，引入transaction coordinator用於記錄transaction log。在叢集中會有多個transaction coordinator，每個tid對應唯一一個transaction coordinator。
註：transaction log刪除策略是compact，已完成的事務會標記成null，compact後不保留。
啟動事務時，先標記開啟事務，寫入資料，全部成功就在transaction log中記錄為prepare commit狀態，否則寫入prepare abort的狀態。之後再去給每個相關的partition寫入一條marker（commit或者abort）訊息，標記這個事務的message可以被讀取或已經廢棄。成功後     在transaction log記錄下commit/abort狀態，至此事務結束。

   整體的資料流是這樣的：

   

1. 首先使用tid請求任意一個broker（程式碼中寫的是負載最小的broker），找到對應的transaction coordinator。
2. 請求transaction coordinator獲取到對應的pid，和pid對應的epoch，這個epoch用於防止僵死行程複活導致訊息錯亂，當訊息的epoch比當前維護的epoch小時，拒絕掉。tid和pid有一一對應的關係，這樣對於同一個tid會傳回相同的pid。
3. client先請求transaction coordinator記錄的事務狀態，初始狀態是BEGIN，如果是該事務中第一個到達的，同時會對事務進行計時；client輸出資料到相關的partition中；client再請求transaction coordinator記錄offset的事務狀態；client傳送offset commit到對應offset partition。
4. client傳送commit請求，transaction coordinator記錄prepare commit/abort，然後傳送marker給相關的partition。全部成功後，記錄commit/abort的狀態，最後這個記錄不需要等待其他replica的ack，因為prepare不丟就能保證最終的正確性了。

     這裡prepare的狀態主要是用於事務恢復，例如給相關的partition傳送控制訊息，沒發完就宕機了，備機起來後，producer傳送請求獲取pid時，會把未完成的事務接著完成。

     當partition中寫入commit的marker後，相關的訊息就可被讀取。所以kafka事務在prepare commit到commit這個時間段內，訊息是逐漸可見的，而不是同一時刻可見。

    訊息消費事務

    消費時，partition中會存在一些訊息處於未commit狀態，即業務方應該看不到的訊息，需要過濾這些訊息不讓業務看到，kafka選擇在消費者行程中進行過來，而不是在broker中過濾，主要考慮的還是效能。kafka高效能的一個關鍵點是zero copy，如果需要在broker中過 濾，那麼勢必需要讀取訊息內容到記憶體，就會失去zero copy的特性。

  五、 Kafka檔案組織

  kafka的資料，實際上是以檔案的形式儲存在檔案系統的。topic下有partition，partition下有segment，

segment是實際的一個個檔案

，topic和partition都是抽象概念。

  在目錄/${topicName}-{$partitionid}/下，儲存著實際的log檔案（即segment），還有對應的索引檔案。

  每個segment檔案大小相等，檔案名以這個segment中最小的offset命名，檔案副檔名是.log；segment對應的索引的檔案名字一樣，副檔名是.index。有兩個index檔案，一個是offset index用於按offset去查message，一個是time index用於按照時間去查，其實這裡可以優化合到一起，下麵只說offset index。總體的組織是這樣的：

  為了減少索引檔案的大小，降低空間使用，方便直接載入進記憶體中，這裡的索引使用稀疏矩陣，不會每一個message都記錄下具體位置，而是每隔一定的位元組數，再建立一條索引。 索引包含兩部分，分別是baseOffset，還有position。

  baseOffset：意思是這條索引對應segment檔案中的第幾條message。這樣做方便使用數值壓縮演演算法來節省空間。例如kafka使用的是varint。

  position：在segment中的絕對位置。

  查詢offset對應的記錄時，會先用二分法，找出對應的offset在哪個segment中，然後使用索引，在定位出offset在segment中的大概位置，再遍歷查詢message。

六、Kafka常用配置項

  Broker配置

  Topic配置

  參考連結：123archu 連結：https://www.jianshu.com/p/d3e963ff8b70