Kafka offset管理

[TOC]
Kafka中的每個partition都由一系列有序的趟薄、不可變的消息組成,這些消息被連續(xù)的追加到partition中呈枉。partition中的每個消息都有一個連續(xù)的序號挂签,用于partition唯一標(biāo)識一條消息谈为。

Offset記錄著下一條將要發(fā)送給Consumer的消息的序號。

Offset從語義上來看擁有兩種:Current Offset和Committed Offset瓤逼。

Current Offset

Current Offset保存在Consumer客戶端中笼吟,它表示Consumer希望收到的下一條消息的序號。它僅僅在poll()方法中使用霸旗。例如贷帮,Consumer第一次調(diào)用poll()方法后收到了20條消息,那么Current Offset就被設(shè)置為20诱告。這樣Consumer下一次調(diào)用poll()方法時撵枢,Kafka就知道應(yīng)該從序號為21的消息開始讀取。這樣就能夠保證每次Consumer poll消息時精居,都能夠收到不重復(fù)的消息诲侮。

Committed Offset

Committed Offset保存在Broker上,它表示Consumer已經(jīng)確認(rèn)消費過的消息的序號箱蟆。主要通過commitSync和commitAsyncAPI來操作沟绪。舉個例子,Consumer通過poll() 方法收到20條消息后空猜,此時Current Offset就是20绽慈,經(jīng)過一系列的邏輯處理后,并沒有調(diào)用consumer.commitAsync()或consumer.commitSync()來提交Committed Offset辈毯,那么此時Committed Offset依舊是0坝疼。

Committed Offset主要用于Consumer Rebalance。在Consumer Rebalance的過程中谆沃,一個partition被分配給了一個Consumer钝凶,那么這個Consumer該從什么位置開始消費消息呢?答案就是Committed Offset唁影。另外耕陷,如果一個Consumer消費了5條消息(poll并且成功commitSync)之后宕機(jī)了,重新啟動之后它仍然能夠從第6條消息開始消費据沈,因為Committed Offset已經(jīng)被Kafka記錄為5哟沫。

總結(jié)一下,Current Offset是針對Consumer的poll過程的锌介,它可以保證每次poll都返回不重復(fù)的消息嗜诀;而Committed Offset是用于Consumer Rebalance過程的猾警,它能夠保證新的Consumer能夠從正確的位置開始消費一個partition,從而避免重復(fù)消費隆敢。

在Kafka 0.9前发皿,Committed Offset信息保存在zookeeper的[consumers/{group}/offsets/{topic}/{partition}]目錄中(zookeeper其實并不適合進(jìn)行大批量的讀寫操作,尤其是寫操作)拂蝎。而在0.9之后雳窟,所有的offset信息都保存在了Broker上的一個名為__consumer_offsets的topic中。

Kafka集群中offset的管理都是由Group Coordinator中的Offset Manager完成的匣屡。

Group Coordinator

Group Coordinator是運行在Kafka集群中每一個Broker內(nèi)的一個進(jìn)程封救。它主要負(fù)責(zé)Consumer Group的管理,Offset位移管理以及Consumer Rebalance捣作。

對于每一個Consumer Group誉结,Group Coordinator都會存儲以下信息:

  1. 訂閱的topics列表
  2. Consumer Group配置信息,包括session timeout等
  3. 組中每個Consumer的元數(shù)據(jù)券躁。包括主機(jī)名惩坑,consumer id
  4. 每個Group正在消費的topic partition的當(dāng)前offsets
  5. Partition的ownership元數(shù)據(jù),包括consumer消費的partitions映射關(guān)系

Consumer Group如何確定自己的coordinator是誰呢也拜? 簡單來說分為兩步:

確定Consumer Group offset信息將要寫入__consumers_offsets topic的哪個分區(qū)以舒。具體計算公式:

__consumers_offsets partition# = Math.abs(groupId.hashCode() % offsets.topic.num.partitions)  //offsets.topic.num.partitions默認(rèn)值為50。


該分區(qū)leader所在的broker就是被選定的coordinator

Offset存儲模型

由于一個partition只能固定的交給一個消費者組中的一個消費者消費慢哈,因此Kafka保存offset時并不直接為每個消費者保存蔓钟,而是以groupid-topic-partition -> offset的方式保存。如圖所示:

image.png

Kafka在保存Offset的時候卵贱,實際上是將Consumer Group和partition對應(yīng)的offset以消息的方式保存在__consumers_offsets這個topic中滥沫。

__consumers_offsets默認(rèn)擁有50個partition,可以通過

Math.abs(groupId.hashCode() % offsets.topic.num.partitions) 

的方式來查詢某個Consumer Group的offset信息保存在__consumers_offsets的哪個partition中键俱。下圖展示了__consumers_offsets中保存的offset消息的格式:

image.png

如圖所示兰绣,一條offset消息的格式為groupid-topic-partition -> offset。因此consumer poll消息時编振,已知groupid和topic缀辩,又通過Coordinator分配partition的方式獲得了對應(yīng)的partition,自然能夠通過Coordinator查找__consumers_offsets的方式獲得最新的offset了踪央。

Offset查詢

前面我們已經(jīng)描述過offset的存儲模型臀玄,它是按照groupid-topic-partition -> offset的方式存儲的。然而Kafka只提供了根據(jù)offset讀取消息的模型杯瞻,并不支持根據(jù)key讀取消息的方式镐牺。那么Kafka是如何支持Offset的查詢呢?

答案就是Offsets Cache?颉!


image.png

如圖所示,Consumer提交offset時旗唁,Kafka Offset Manager會首先追加一條條新的commit消息到__consumers_offsets topic中畦浓,然后更新對應(yīng)的緩存。讀取offset時從緩存中讀取检疫,而不是直接讀取__consumers_offsets這個topic讶请。

Log Compaction

我們已經(jīng)知道,Kafka使用groupid-topic-partition -> offset*的消息格式屎媳,將Offset信息存儲在__consumers_offsets topic中夺溢。請看下面一個例子:

image.png

如圖,對于audit-consumer這個Consumer Group來說烛谊,上面的存儲了兩條具有相同key的記錄:PageViewEvent-0 -> 240和PageViewEvent-0 -> 323风响。事實上,這就是一種無用的冗余丹禀。因為對于一個partition來說状勤,我們實際上只需要它當(dāng)前最新的Offsets。

為了消除這樣的過期數(shù)據(jù)双泪,Kafka為__consumers_offsets topic設(shè)置了Log Compaction功能持搜。Log Compaction意味著對于有相同key的的不同value值,只保留最后一個版本焙矛。如果應(yīng)用只關(guān)心key對應(yīng)的最新value值葫盼,可以開啟Kafka的Log Compaction功能,Kafka會定期將相同key的消息進(jìn)行合并村斟,只保留最新的value值剪返。

image.png

在新建topic時添加log.cleanup.policy=compact參數(shù)就可以為topic開啟Log Compaction功能。

auto.offset.reset參數(shù)

auto.offset.reset表示如果Kafka中沒有存儲對應(yīng)的offset信息的話(有可能offset信息被刪除)邓梅,消費者從何處開始消費消息脱盲。它擁有三個可選值:

  • earliest:從最早的offset開始消費
  • latest:從最后的offset開始消費
  • none:直接拋出exception給consumer

看一下下面兩個場景:

  1. Consumer消費了5條消息后宕機(jī)了,重啟之后它讀取到對應(yīng)的partition的Committed Offset為5日缨,因此會直接從第6條消息開始讀取钱反。此時完全依賴于Committed Offset機(jī)制,和auto.offset.reset配置完全無關(guān)匣距。
  2. 新建了一個新的Group面哥,并添加了一個Consumer,它訂閱了一個已經(jīng)存在的Topic毅待。此時Kafka中還沒有這個Consumer相應(yīng)的Offset信息尚卫,因此此時Kafka就會根據(jù)auto.offset.reset配置來決定這個Consumer從何處開始消費消息。
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末尸红,一起剝皮案震驚了整個濱河市吱涉,隨后出現(xiàn)的幾起案子刹泄,更是在濱河造成了極大的恐慌,老刑警劉巖怎爵,帶你破解...
    沈念sama閱讀 222,590評論 6 517
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件特石,死亡現(xiàn)場離奇詭異,居然都是意外死亡鳖链,警方通過查閱死者的電腦和手機(jī)姆蘸,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,157評論 3 399
  • 文/潘曉璐 我一進(jìn)店門,熙熙樓的掌柜王于貴愁眉苦臉地迎上來芙委,“玉大人逞敷,你說我怎么就攤上這事」嗦拢” “怎么了推捐?”我有些...
    開封第一講書人閱讀 169,301評論 0 362
  • 文/不壞的土叔 我叫張陵,是天一觀的道長顶瞳。 經(jīng)常有香客問我玖姑,道長,這世上最難降的妖魔是什么慨菱? 我笑而不...
    開封第一講書人閱讀 60,078評論 1 300
  • 正文 為了忘掉前任焰络,我火速辦了婚禮,結(jié)果婚禮上符喝,老公的妹妹穿的比我還像新娘闪彼。我一直安慰自己,他們只是感情好协饲,可當(dāng)我...
    茶點故事閱讀 69,082評論 6 398
  • 文/花漫 我一把揭開白布畏腕。 她就那樣靜靜地躺著,像睡著了一般茉稠。 火紅的嫁衣襯著肌膚如雪描馅。 梳的紋絲不亂的頭發(fā)上,一...
    開封第一講書人閱讀 52,682評論 1 312
  • 那天而线,我揣著相機(jī)與錄音铭污,去河邊找鬼。 笑死膀篮,一個胖子當(dāng)著我的面吹牛嘹狞,可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播誓竿,決...
    沈念sama閱讀 41,155評論 3 422
  • 文/蒼蘭香墨 我猛地睜開眼磅网,長吁一口氣:“原來是場噩夢啊……” “哼!你這毒婦竟也來了筷屡?” 一聲冷哼從身側(cè)響起涧偷,我...
    開封第一講書人閱讀 40,098評論 0 277
  • 序言:老撾萬榮一對情侶失蹤簸喂,失蹤者是張志新(化名)和其女友劉穎,沒想到半個月后嫂丙,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體娘赴,經(jīng)...
    沈念sama閱讀 46,638評論 1 319
  • 正文 獨居荒郊野嶺守林人離奇死亡规哲,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點故事閱讀 38,701評論 3 342
  • 正文 我和宋清朗相戀三年跟啤,在試婚紗的時候發(fā)現(xiàn)自己被綠了。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片唉锌。...
    茶點故事閱讀 40,852評論 1 353
  • 序言:一個原本活蹦亂跳的男人離奇死亡隅肥,死狀恐怖,靈堂內(nèi)的尸體忽然破棺而出袄简,到底是詐尸還是另有隱情腥放,我是刑警寧澤,帶...
    沈念sama閱讀 36,520評論 5 351
  • 正文 年R本政府宣布绿语,位于F島的核電站秃症,受9級特大地震影響,放射性物質(zhì)發(fā)生泄漏吕粹。R本人自食惡果不足惜种柑,卻給世界環(huán)境...
    茶點故事閱讀 42,181評論 3 335
  • 文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望匹耕。 院中可真熱鬧聚请,春花似錦、人聲如沸稳其。這莊子的主人今日做“春日...
    開封第一講書人閱讀 32,674評論 0 25
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽既鞠。三九已至煤傍,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間嘱蛋,已是汗流浹背蚯姆。 一陣腳步聲響...
    開封第一講書人閱讀 33,788評論 1 274
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機(jī)就差點兒被人妖公主榨干…… 1. 我叫王不留浑槽,地道東北人蒋失。 一個月前我還...
    沈念sama閱讀 49,279評論 3 379
  • 正文 我出身青樓,卻偏偏與公主長得像桐玻,于是被迫代替她去往敵國和親篙挽。 傳聞我的和親對象是個殘疾皇子,可洞房花燭夜當(dāng)晚...
    茶點故事閱讀 45,851評論 2 361