2019獨(dú)角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>

我們先假設(shè)初始時(shí)世界是混沌的還沒(méi)有盤(pán)古的開(kāi)天辟地，協(xié)調(diào)者也是一片荒蕪人煙之地，沒(méi)有保存任何狀態(tài)，因?yàn)橄M(fèi)組的初始狀態(tài)是Stable，在第一次的Rebalance時(shí)，正常的還沒(méi)有向消費(fèi)組注冊(cè)過(guò)的消費(fèi)者會(huì)執(zhí)行狀態(tài)為Stable而且memberId=UNKNOWN_MEMBER_ID條件分支。在第一次Rebalance之后，每個(gè)消費(fèi)者都分配到了一個(gè)成員編號(hào)，系統(tǒng)又會(huì)進(jìn)入Stable穩(wěn)定狀態(tài)（Stable穩(wěn)定狀態(tài)包括兩種：一種是沒(méi)有任何消費(fèi)者的穩(wěn)定狀態(tài)，一種是有消費(fèi)者的穩(wěn)定狀態(tài)）。因?yàn)樗邢M(fèi)者在執(zhí)行一次JoinGroup后并不是說(shuō)系統(tǒng)就一直保持這種不變的狀態(tài)，有可能因?yàn)檫@樣或那樣的事件導(dǎo)致消費(fèi)者要重新進(jìn)行JoinGroup，這個(gè)時(shí)候因?yàn)橹癑oinGroup過(guò)了每個(gè)消費(fèi)者都是有成員編號(hào)的，處理方式肯定是不一樣的。

所以定義一種事件驅(qū)動(dòng)的狀態(tài)機(jī)就很有必要了，這世界看起來(lái)是雜亂無(wú)章的，不過(guò)只要遵循著狀態(tài)機(jī)的規(guī)則（萬(wàn)物生長(zhǎng)的理論），任何事件都是有跡可循有路可走有條不紊地進(jìn)行著。

?

private def doJoinGroup(group: GroupMetadata,memberId: String,clientId: String,clientHost: String,sessionTimeoutMs: Int,protocolType: String,protocols: List[(String, Array[Byte])],responseCallback: JoinCallback) {if (group.protocolType!=protocolType||!group.supportsProtocols(protocols.map(_._1).toSet)) {//protocolType對(duì)于消費(fèi)者是consumer，注意這里的協(xié)議類型和PartitionAssignor協(xié)議不同哦//協(xié)議類型目前總共就兩種消費(fèi)者和Worker，而協(xié)議是PartitionAssignor分配算法responseCallback(joinError(memberId, Errors.INCONSISTENT_GROUP_PROTOCOL.code))} else if (memberId != JoinGroupRequest.UNKNOWN_MEMBER_ID && !group.has(memberId)) {//如果當(dāng)前組沒(méi)有記錄該消費(fèi)者，而該消費(fèi)者卻被分配了成員編號(hào)，則重置為未知成員，并讓消費(fèi)者重試responseCallback(joinError(memberId, Errors.UNKNOWN_MEMBER_ID.code))} else { group.currentState match {case Dead =>responseCallback(joinError(memberId, Errors.UNKNOWN_MEMBER_ID.code))case PreparingRebalance =>if (memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID) { //2.第二個(gè)消費(fèi)者在這里了！addMemberAndRebalance(sessionTimeoutMs, clientId, clientHost, protocols, group, responseCallback)} else {val member = group.get(memberId)updateMemberAndRebalance(group, member, protocols, responseCallback)}case Stable =>if (memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID) { //1.初始時(shí)第一個(gè)消費(fèi)者在這里！//如果消費(fèi)者成員編號(hào)是未知的，則向GroupMetadata注冊(cè)并被記錄下來(lái)addMemberAndRebalance(sessionTimeoutMs, clientId, clientHost, protocols, group, responseCallback)} else { //3.第二次Rebalance時(shí)第一個(gè)消費(fèi)者在這里，此時(shí)要分Leader還是普通的消費(fèi)者了val member = group.get(memberId)if (memberId == group.leaderId || !member.matches(protocols)) {updateMemberAndRebalance(group, member, protocols, responseCallback)} else {responseCallback(JoinGroupResult(members = Map.empty,memberId = memberId,generationId = group.generationId,subProtocol = group.protocol,leaderId = group.leaderId,errorCode = Errors.NONE.code))}}}if (group.is(PreparingRebalance))joinPurgatory.checkAndComplete(GroupKey(group.groupId))} }

addMemberAndRebalance和updateMemberAndRebalance會(huì)創(chuàng)建或更新MemberMetadata，并且會(huì)嘗試調(diào)用prepareRebalance，消費(fèi)組中只有一個(gè)消費(fèi)者有機(jī)會(huì)調(diào)用prepareRebalance，并且一旦調(diào)用該方法，會(huì)將消費(fèi)組狀態(tài)更改為PreparingRebalance，就會(huì)使得下一個(gè)消費(fèi)者只能從case PreparingRebalance入口進(jìn)去了，假設(shè)第一個(gè)消費(fèi)者是從Stable進(jìn)入的，它更改了狀態(tài)為PreparingRebalance，下一個(gè)消費(fèi)者就不會(huì)從Stable進(jìn)來(lái)的。不過(guò)進(jìn)入Stable狀態(tài)還要判斷消費(fèi)者是不是已經(jīng)有了成員編號(hào)，通常是之前已經(jīng)發(fā)生了Rebalance，這種影響也是比較巨大的，每個(gè)消費(fèi)者走的路徑跟第一次的Rebalance是完全不同的迷宮地圖了。

1）第一次Rebalance如圖6-18的上半部分：

第一個(gè)消費(fèi)者，狀態(tài)為Stable，沒(méi)有編號(hào)，addMemberAndRebalance，成為L(zhǎng)eader，執(zhí)行prepareRebalance，更改狀態(tài)為PreparingRebalance，創(chuàng)建DelayedJoin

第二個(gè)消費(fèi)者，狀態(tài)為PreparingRebalance，沒(méi)有編號(hào)，addMemberAndRebalance（不執(zhí)行prepareRebalance，因?yàn)樵跔顟B(tài)改變成PreparingRebalance后就不會(huì)被執(zhí)行了）；后面的消費(fèi)者同第二個(gè)

所有消費(fèi)者都要等協(xié)調(diào)者收集完所有成員編號(hào)在DelayedJoin完成時(shí)才會(huì)收到JoinGroup響應(yīng)

?

圖6-18 第一次和第二次Rebalance

2）第二次Rebalance，對(duì)于之前加入過(guò)的消費(fèi)者都要成員編號(hào)如圖6-18的下半部分：

第一個(gè)消費(fèi)者是Leader，狀態(tài)為Stable，有編號(hào)，updateMemberAndRebalance，更改狀態(tài)為PreparingRebalance，創(chuàng)建DelayedJoin

第二個(gè)消費(fèi)者，狀態(tài)為PreparingRebalance，有編號(hào)，updateMemberAndRebalance；后面的消費(fèi)者同第二個(gè)

所有消費(fèi)者也要等待，因?yàn)槠渌M(fèi)者發(fā)送Join請(qǐng)求在Leader消費(fèi)者之后。

3）不過(guò)如果有消費(fèi)者在Leader之前發(fā)送又有點(diǎn)不一樣了如圖6-19：

第一個(gè)消費(fèi)者不是Leader，狀態(tài)為Stable，有編號(hào)，responseCallback，立即收到JoinGroup響應(yīng)，好幸運(yùn)啊！

第二個(gè)消費(fèi)者如果也不是Leader，恭喜你，協(xié)調(diào)者也放過(guò)他，直接返回JoinGroup響應(yīng)

第三個(gè)消費(fèi)者是Leader（領(lǐng)導(dǎo)來(lái)了），狀態(tài)為Stable（什么，你們之前的消費(fèi)者竟然都沒(méi)更新?tīng)顟B(tài)！，因?yàn)樗麄兌紱](méi)有add或update），有編號(hào)，updateMemberAndRebalance（還是我第一個(gè)調(diào)用add或update，看來(lái)還是只能我來(lái)更新?tīng)顟B(tài)），更改狀態(tài)為PreparingRebalance，創(chuàng)建DelayedJoin

第四個(gè)消費(fèi)者不是Leader，狀態(tài)為PreparingRebalance，有編號(hào)，updateMemberAndRebalance（前面有領(lǐng)導(dǎo)，不好意思了，不能立即返回JoinGroup給你了，你們這些剩下的消費(fèi)者都只能和領(lǐng)導(dǎo)一起返回了，算你們倒霉）

?

圖6-19 Leader非第一個(gè)發(fā)送JoinGroup請(qǐng)求

4）如果第一個(gè)消費(fèi)者不是Leader，也沒(méi)有編號(hào)，說(shuō)明這是一個(gè)新增的消費(fèi)者，流程又不同了如圖6-20：

第一個(gè)消費(fèi)者不是Leader，狀態(tài)為Stable，沒(méi)有編號(hào)，addMemberAndRebalance，執(zhí)行prepareRebalance（我是第一個(gè)調(diào)用add或update的哦，你們都別想跟我搶這個(gè)頭彩了），更改狀態(tài)為PreparingRebalance（我不是Leader但我驕傲啊），創(chuàng)建DelayedJoin（我搶到頭彩，當(dāng)然創(chuàng)建DelayedJoin的工作只能由我來(lái)完成了）

第二個(gè)消費(fèi)者也不是Leader，恭喜你，協(xié)調(diào)者也放過(guò)他，直接返回JoinGroup響應(yīng)

第三個(gè)消費(fèi)者是Leader（領(lǐng)導(dǎo)來(lái)了），狀態(tài)為PreparingRebalance（有個(gè)新來(lái)的不懂規(guī)矩，他已經(jīng)把狀態(tài)改了），有編號(hào)，updateMemberAndRebalance（有人已經(jīng)改了，你老就不用費(fèi)心思了），凡是沒(méi)有立即返回響應(yīng)的，都需要等待，領(lǐng)導(dǎo)也不例外

第四個(gè)消費(fèi)者不是Leader（廢話，只有一個(gè)領(lǐng)導(dǎo)，而且領(lǐng)導(dǎo)已經(jīng)在前面了），不會(huì)立即返回響應(yīng)（你看領(lǐng)導(dǎo)都排隊(duì)呢）

雖然DelayedJoin是由沒(méi)有編號(hào)的消費(fèi)者創(chuàng)建，不過(guò)由于DelayedJoin是以消費(fèi)組為級(jí)別的，所以不用擔(dān)心，上一次選舉出來(lái)的領(lǐng)導(dǎo)還是領(lǐng)導(dǎo)，協(xié)調(diào)者最終還是會(huì)把members交給領(lǐng)導(dǎo)，不會(huì)是給那個(gè)沒(méi)有編號(hào)的消費(fèi)者的，雖然說(shuō)在他注冊(cè)的時(shí)候已經(jīng)有編號(hào)了，但是大家不認(rèn)啊。不過(guò)領(lǐng)導(dǎo)其實(shí)不在意是誰(shuí)開(kāi)始觸發(fā)prepareRebalance的，那個(gè)人要負(fù)責(zé)生成DelayedJoin，而不管是領(lǐng)導(dǎo)自己還是其他人一旦更改狀態(tài)為PreparingRebalance，后面的消費(fèi)者都要等待DelayedJoin完成了，而領(lǐng)導(dǎo)者總是要等待的，所以他當(dāng)然無(wú)所謂了，因?yàn)樗雷詈髤f(xié)調(diào)者總是會(huì)把members交給他的。

?

圖6-20 新增消費(fèi)組第一個(gè)發(fā)送JoinGroup請(qǐng)求

根據(jù)上面的幾種場(chǎng)景總結(jié)下來(lái)狀態(tài)機(jī)的規(guī)則和一些結(jié)論如下：

第一個(gè)調(diào)用addMemberAndRebalance或者updateMemberAndRebalance的會(huì)將狀態(tài)改為PreparingRebalance，并且負(fù)責(zé)生成DelayedJoin

一旦狀態(tài)進(jìn)入PreparingRebalance，其他消費(fèi)者就只能從PreparingRebalance狀態(tài)入口進(jìn)入，這里只有兩種選擇addMemberAndRebalance或者updateMemberAndRebalance，不過(guò)他們不會(huì)更改狀態(tài)，也不會(huì)生成DelayedJoin

發(fā)生DelayedJoin之后，其他消費(fèi)者的JoinGroup響應(yīng)都會(huì)被延遲，因?yàn)槿缫?guī)則2中，他們只能調(diào)用add或update，無(wú)法立即調(diào)用responseCallback，所以就要和DelayedJoin的那個(gè)消費(fèi)者一起等待

正常流程時(shí)，發(fā)生responseCallback的是存在成員編號(hào)的消費(fèi)者在Leader之前發(fā)送了JoinGroup，或者新增加的消費(fèi)者發(fā)送了JoinGroup請(qǐng)求之前

第一次Rebalance時(shí)，第一個(gè)消費(fèi)者會(huì)創(chuàng)建DelayedJoin，之后的Rebalance，只有新增的消費(fèi)者才有機(jī)會(huì)創(chuàng)建（如果他在Leader之前發(fā)送的話，如果在Leader之后就沒(méi)有機(jī)會(huì)了），而普通消費(fèi)者總是沒(méi)有機(jī)會(huì)創(chuàng)建DelayedJoin的，因?yàn)闋顟B(tài)為Stable時(shí)，他會(huì)直接開(kāi)溜，有人（Leader或者新增加的消費(fèi)者）創(chuàng)建了DelayedJoin之后，他又在那邊怨天尤人只能等待

轉(zhuǎn)載于:https://my.oschina.net/u/2371517/blog/1142949

總結(jié)

以上是生活随笔為你收集整理的kafka协调者的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。