• Kafka的消息消費(fèi)是以消費(fèi)的group為單位;
• 同屬一個(gè)group中的多個(gè)consumer分別消費(fèi)topic的不同partition;
• 同組內(nèi)consumer的變化, partition變化, coordinator的變化都會(huì)引發(fā)balance;
• 消費(fèi)的offset的提交
• Kafka wiki: Kafka Detailed Consumer Coordinator Design 和 Kafka Client-side Assignment Proposal

GroupMetadata類

• 所在文件: core/src/main/scala/kafka/coordinator/MemberMetadata.scala
• 作用: 用來(lái)表示一個(gè)消費(fèi)group的相關(guān)信息
• 當(dāng)前group的狀態(tài): private var state: GroupState = Stable
  1. Stable: consumer group的balance已完成, 處于穩(wěn)定狀態(tài);
  2. PreparingRebalance: 收到JoinRequest, consumer group需要重新作balance時(shí)的狀態(tài);
  3. AwaitingSync: 收到了所有需要的JoonRequest, 等待作為當(dāng)前group的leader的consumer客戶端提交balance的結(jié)果到coordinator;
  4. Dead: 當(dāng)前的消費(fèi)group不再有任何consumer成員時(shí)的狀態(tài);
• 當(dāng)前group的成員相關(guān)信息:
  1. 成員信息: private val members = new mutable.HashMap[String, MemberMetadata],
     每個(gè)成員都有一個(gè)memberId, 對(duì)應(yīng)著MemberMetadata;
  2. var leaderId: String: 對(duì)于group的balance, 簡(jiǎn)單來(lái)講實(shí)際上是Coordinator收集了所有的consumer的信息后, 將其發(fā)送給group中的一個(gè)consumer, 這個(gè)consumer負(fù)責(zé)按一定的balance策略,將partition分配到不同的consumer, 這個(gè)分配結(jié)果會(huì)Sync回Coordinator, 然后再同步到各個(gè)consumer, 這個(gè)負(fù)責(zé)具體分配的consumer就是當(dāng)前的Leader; 這個(gè)Leader的決定很簡(jiǎn)單, 誰(shuí)第一個(gè)加入這個(gè)group的,誰(shuí)就是leader;
  3. var protocol: String: 當(dāng)前group組所采用的balance策略, 選取的規(guī)則是被當(dāng)前所有member都支持的策略中最多的那一個(gè);
  4. var generationId: 當(dāng)前balance的一個(gè)標(biāo)識(shí)id, 可以簡(jiǎn)單理解成是第幾次作balance, 每次狀態(tài)轉(zhuǎn)換到AwaitingSync時(shí), 其值就增加1;

GroupMetadataManager類

• 所在文件: core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
• 作用: 是比較核心的一個(gè)類, 負(fù)責(zé)所有g(shù)roup的管理, offset消息的讀寫和清理等, 下面我們一一道來(lái)
• 當(dāng)前所有消費(fèi)group的管理:
  1. private val groupsCache = new Pool[String, GroupMetadata]: 緩存了所有GroupMetadata的信息;
  2. 針對(duì)groupsCache的管理接口:

def getGroup(groupId: String): GroupMetadata
def addGroup(group: GroupMetadata): GroupMetadata
def removeGroup(group: GroupMetadata)

• __consumer_offsets topic的讀寫
  1. 我們已經(jīng)知道現(xiàn)在的kafka已經(jīng)支持將offset信息保存到broker上, 實(shí)際上是保存到一個(gè)內(nèi)部的topic上:__consumer_offsets, 寫入其中的msg都包含有key
  2. __consumer_offsets這個(gè)topic里實(shí)際上保存兩種類型消息:
     2.1 一部分是offset信息(kafka.coordinator.OffsetsMessageFormatter類型)的：
     [groupId,topic,partition]::[OffsetMetadata[offset,metadata],CommitTime ExprirationTime], 它的key是 [groupId,topic,partition]
     2.2 另一部分是group信息(kafka.coordinator.GroupMetadataMessageFormatter類型):
     groupId::[groupId,Some(consumer),groupState,Map(memberId -> [memberId,clientId,clientHost,sessionTimeoutMs], ...->[]...)], 這部分實(shí)際上就是把當(dāng)前Stable狀態(tài)的GroupMetadata存到了__consumer_offsets里, , 它的key是 groupId
  3. offset和group信息的寫入: 實(shí)際上是普通的消息寫入沒(méi)有本質(zhì)上的區(qū)別, 可參考Kafka是如何處理客戶端發(fā)送的數(shù)據(jù)的？, 這里的方法是def store(delayedAppend: DelayedStore), 實(shí)現(xiàn)就是調(diào)用replicaManager.appendMessages來(lái)寫入消息到log文件
• __consumer_offsets topic消息的加載
  1. __consumer_offsets作為一個(gè)topic, 也是有多個(gè)partiton的, 每個(gè)partiton也是有多個(gè)復(fù)本的, partition也會(huì)經(jīng)歷leader的選舉,也會(huì)有故障轉(zhuǎn)移操作;
  2. 當(dāng)__consumer_offsets在某臺(tái)broker上的partition成為leader partition時(shí), 需要先從本地的log文件后加載offset,group相關(guān)信息到內(nèi)存, 加載完成后才能對(duì)外提供讀寫和balance的操作;
  3. 具體實(shí)現(xiàn): def loadGroupsForPartition(offsetsPartition: Int, onGroupLoaded: GroupMetadata => Unit)
• offset的相關(guān)操作
  1. 使用者消費(fèi)msg提交的offset, 不僅會(huì)寫入到log文件后, 為了快速響應(yīng)還會(huì)緩存在內(nèi)存中, 對(duì)應(yīng)private val offsetsCache = new Pool[GroupTopicPartition, OffsetAndMetadata];
  2. 直接從內(nèi)存中獲取某一group對(duì)應(yīng)某一topic的parition的offset信息:
     def getOffsets(group: String, topicPartitions: Seq[TopicAndPartition]): Map[TopicAndPartition, OffsetMetadataAndError]
  3. 刷新offset: offsetsCache只保存最后一次提交的offset信息
     private def putOffset(key: GroupTopicPartition, offsetAndMetadata: OffsetAndMetadata)
• 刪除過(guò)期的offset消息
  1. GroupMetadataManager在啟動(dòng)時(shí)會(huì)同時(shí)啟動(dòng)一個(gè)名為delete-expired-consumer-offsets定時(shí)任務(wù)來(lái)定時(shí)刪除過(guò)期的offset信息;
  2. 從內(nèi)存緩存中清除: offsetsCache.remove(groupTopicAndPartition)
  3. 從已經(jīng)落地的log文件中清除: 實(shí)現(xiàn)就是向log里寫一條payload為null的"墓碑"message作為標(biāo)記, __consumer_offsets的清除策略默認(rèn)是compact, 后面我們會(huì)單獨(dú)開(kāi)一章來(lái)講日志的清除;

GroupCoordinator類

• 所在文件: core/src/main/scala/kafka/coordinator/GroupCoordinator.scala
• 核心類, 處理所有和消息消費(fèi)相關(guān)的request:

       case RequestKeys.OffsetCommitKey => handleOffsetCommitRequest(request)
       case RequestKeys.OffsetFetchKey => handleOffsetFetchRequest(request)
       case RequestKeys.GroupCoordinatorKey => handleGroupCoordinatorRequest(request)
       case RequestKeys.JoinGroupKey => handleJoinGroupRequest(request)
       case RequestKeys.HeartbeatKey => handleHeartbeatRequest(request)
       case RequestKeys.LeaveGroupKey => handleLeaveGroupRequest(request)
       case RequestKeys.SyncGroupKey => handleSyncGroupRequest(request)
       case RequestKeys.DescribeGroupsKey => handleDescribeGroupRequest(request)
       case RequestKeys.ListGroupsKey => handleListGroupsRequest(request)

• 使用簡(jiǎn)單狀態(tài)機(jī)來(lái)協(xié)調(diào)consumer group的balance;
• 下面我們假設(shè)在一個(gè)group:g1中啟動(dòng)兩個(gè)consumer: c1和c2來(lái)消費(fèi)同一個(gè)topic, 來(lái)看看狀態(tài)機(jī)的轉(zhuǎn)換
  1. 第一種情況: c1和c2分別啟動(dòng):

c2.jpg

2. 第二種情況: c1和c2已經(jīng)在group中, 然后c1正常的退出離開(kāi)

c1.jpg

3. 第二種情況: c1和c2已經(jīng)在group中, 然后c1非正常退出,比如說(shuō)進(jìn)程被kill掉
   流程跟上面的2基本上一致, 只不過(guò)(1)這步的觸發(fā)條件不是LeaveGroupRequest, 而是來(lái)自c1的heartbeat的onExpireHeartbeat;
4. 第四種情況: c1和c2已經(jīng)在group中, 然后這個(gè)topic的partition增加, 這個(gè)時(shí)候服務(wù)端是無(wú)法主動(dòng)觸發(fā)的,客戶端會(huì)定時(shí)去服務(wù)端同步metadata信息, 從新的metadata信息中客戶端會(huì)獲知partition有了變化, 此時(shí)c1和c2會(huì)重新發(fā)送JoinRequest來(lái)觸發(fā)新的balance;
5. 還有其它的兩種情況, 這里就不一一說(shuō)明了,總之就是利用這個(gè)狀態(tài)機(jī)的轉(zhuǎn)換來(lái)作相應(yīng)的處理.

Kafka源碼分析-匯總