歡迎光臨
每天分享高質量文章

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

赞(0)

分享創造快樂