- commit offset時(shí)可以附帶一個(gè)string類(lèi)型的metadata用于添加一些有關(guān)信息 也可以附帶一個(gè)long類(lèi)型的timestamp
- 使用完畢需要close 否則會(huì)造成TCP連接的內(nèi)存泄漏
- 消費(fèi)者不是線程安全的
- 調(diào)用老Broker不支持的新特性時(shí) 會(huì)收到UnsupportedVersionException
- 消費(fèi)者偏移量有兩種更新方式
- 隨著每一次調(diào)用poll()方法自動(dòng)前進(jìn)
- 調(diào)用commitSync()/commitAsync()方法主動(dòng)提交偏移量琢岩，默認(rèn)也會(huì)定時(shí)提交(5S)
- 相同group.id的消費(fèi)者在同一group中，這些消費(fèi)者可以部署在同一機(jī)器上，也可以部署在不同機(jī)器上以提供擴(kuò)展性和容錯(cuò)性
- 消費(fèi)者可以動(dòng)態(tài)訂閱主題列表，通過(guò)subsribe()接口
- 一個(gè)消息只會(huì)發(fā)送給一個(gè)組中的一個(gè)消費(fèi)者
- 每個(gè)分區(qū)(partition)屬于一個(gè)消費(fèi)者，分區(qū)會(huì)均衡的分配給所有的消費(fèi)者
- 這種從屬關(guān)系會(huì)被動(dòng)態(tài)的維持割以，當(dāng)原消費(fèi)者關(guān)閉時(shí)，會(huì)分配給其他的消費(fèi)者。同樣氓英，新的消費(fèi)者加入后，會(huì)從舊消費(fèi)者中獲取分區(qū)的所有權(quán)鹦筹。這被稱(chēng)為組重平衡
- 當(dāng)新的分區(qū)被創(chuàng)建铝阐，新的符合訂閱規(guī)則的主題被創(chuàng)建時(shí)，組可以通過(guò)定時(shí)的更新元數(shù)據(jù)來(lái)獲知并觸發(fā)組重平衡
- 概念上 一個(gè)消費(fèi)者群可以看做一個(gè)獨(dú)立的多線程的消費(fèi)者 kafka理所當(dāng)然的支持大量的群組訂閱同一主題
- 在隊(duì)列消息系統(tǒng)中 同一消費(fèi)群的需要類(lèi)似于排隊(duì)式的消費(fèi)這些消息 而發(fā)布訂閱消息系統(tǒng)中 每一個(gè)消費(fèi)者都是單獨(dú)的群組 需要讀取所訂閱主題的所有消息 kafka使用消費(fèi)者組統(tǒng)一了兩種模型
- 消費(fèi)者可以通過(guò)ConsumerRebalanceListener來(lái)監(jiān)聽(tīng)組重平衡的發(fā)生以完成一些收尾的工作
- 消費(fèi)者可以主動(dòng)地選擇自己的分區(qū) assign()接口 這種情況 組內(nèi)的協(xié)調(diào)將會(huì)失效
- 消費(fèi)者的第一次poll()會(huì)建立起對(duì)broker的連接 同時(shí)也會(huì)觸發(fā)組重平衡獲取從屬分區(qū)
- broker通過(guò)poll來(lái)確認(rèn)消費(fèi)者活著铐拐，消費(fèi)者會(huì)定期的發(fā)送心跳包給broker 超過(guò)時(shí)限未收到心跳包 則判斷消費(fèi)者死亡 即心跳包和poll缺一不可
- 區(qū)別 不調(diào)用poll是消費(fèi)者主動(dòng)斷鏈 心跳包是broker主動(dòng)斷鏈
- 僅活躍成員可commit offset 不poll時(shí)commit會(huì)收到CommitFailedException
- max.poll.interval.ms 可以通過(guò)調(diào)整該參數(shù)來(lái)調(diào)整消費(fèi)者用于處理消息的時(shí)間 過(guò)大的增加這個(gè)值會(huì)推遲組重平衡的發(fā)生 如果該值過(guò)小 可能會(huì)導(dǎo)致消費(fèi)者來(lái)不及處理獲取到的消息
- max.poll.records 該參數(shù)表明一次返回的消息的最大值 可以防止消費(fèi)者不能在指定的時(shí)間內(nèi)處理掉所有獲取的消息
- 如果消息處理的時(shí)間不可預(yù)料徘键，我們可以將消息的處理放入單獨(dú)的線程练对，需要注意：確保提交的偏移量是正確的。我們可以通過(guò)關(guān)閉自動(dòng)確認(rèn)偏移量并且在每個(gè)線程結(jié)束后手動(dòng)提交偏移量吹害。在未處理完成前螟凭，可以使用pause()來(lái)停止獲取相應(yīng)分區(qū)的新消息
- bootstrap.servers只需要包含broker集群中的一部分即可
- 當(dāng)消息的消費(fèi)與處理耦合時(shí)，我們需要確認(rèn)消息被處理后手動(dòng)提交偏移量
- 當(dāng)消費(fèi)者保持高可用且失敗后重試時(shí)(cluster management framework or stream processing framework)它呀，就不需要kafka監(jiān)聽(tīng)它的情況了
- 通過(guò)使用非kafka存儲(chǔ)offset 我們可以實(shí)現(xiàn)exactly once 需要關(guān)閉auto.commit 保存每個(gè)record的offset 重啟時(shí)seek到保存的offset
- seekToBeginning seekToEnd
- 可以使用pause()和resume()來(lái)動(dòng)態(tài)的限制對(duì)主題或者分區(qū)的消費(fèi)
- 0.11.0后 kafka加入了對(duì)事務(wù)的支持螺男，將消費(fèi)者的隔離級(jí)別設(shè)置成read_commited，消費(fèi)者僅能讀取那些已經(jīng)被commit的事務(wù)的消息纵穿，這種情況下消費(fèi)者分區(qū)內(nèi)最大偏移位最后一個(gè)完成的事務(wù)的偏移 Last Stable offset (LSO)
- 使用wakeup()來(lái)安全的關(guān)閉消費(fèi)者線程
- 多線程的使用consumer(一個(gè)消費(fèi)者一個(gè)線程)
- (優(yōu)) 易擴(kuò)展 / 一個(gè)消費(fèi)者一個(gè)線程是最佳實(shí)踐下隧，可以免去線程間通信的花費(fèi) / 可以非常容易的實(shí)現(xiàn)分區(qū)的有序處理
- (缺) 每一個(gè)消費(fèi)者對(duì)應(yīng)著一個(gè)TCP連接，kafka通常情況可以高效的管理這些連接谓媒。
- (缺) 大量的消費(fèi)者發(fā)送大量的請(qǐng)求到服務(wù)器淆院，分批次的發(fā)送效果變差，可能會(huì)降低I/O吞吐量
- (缺) 線程總數(shù)收分區(qū)總數(shù)所限制
- 消費(fèi)與處理的解耦 使用一定數(shù)量的消費(fèi)線程消費(fèi)所有數(shù)據(jù)句惯，將數(shù)據(jù)放入一個(gè)阻塞的隊(duì)列中土辩，在使用一個(gè)處理線程池來(lái)真正的處理這些消息
- (優(yōu)) 可以自由確定消費(fèi)者和處理線程的數(shù)量比例，不受分區(qū)數(shù)量的限制
- (缺) 無(wú)法保證處理的先后順序 大部分情況下不是問(wèn)題
- (缺) 主動(dòng)的提交偏移變得非常困難 需要多線程之間的協(xié)調(diào)以確保分區(qū)的消費(fèi)完成
kafka消費(fèi)者 細(xì)節(jié)
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
- 文/潘曉璐 我一進(jìn)店門(mén)详恼，熙熙樓的掌柜王于貴愁眉苦臉地迎上來(lái)补君，“玉大人，你說(shuō)我怎么就攤上這事昧互⊥焯” “怎么了？”我有些...
- 文/不壞的土叔 我叫張陵敞掘，是天一觀的道長(zhǎng)叽掘。 經(jīng)常有香客問(wèn)我，道長(zhǎng)玖雁，這世上最難降的妖魔是什么更扁？ 我笑而不...
- 正文 為了忘掉前任，我火速辦了婚禮，結(jié)果婚禮上浓镜，老公的妹妹穿的比我還像新娘溃列。我一直安慰自己，他們只是感情好膛薛，可當(dāng)我...
- 文/花漫 我一把揭開(kāi)白布听隐。 她就那樣靜靜地躺著，像睡著了一般相叁。 火紅的嫁衣襯著肌膚如雪遵绰。 梳的紋絲不亂的頭發(fā)上，一...
- 那天增淹，我揣著相機(jī)與錄音椿访，去河邊找鬼。 笑死虑润，一個(gè)胖子當(dāng)著我的面吹牛成玫，可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播拳喻，決...
- 文/蒼蘭香墨 我猛地睜開(kāi)眼哭当，長(zhǎng)吁一口氣：“原來(lái)是場(chǎng)噩夢(mèng)啊……” “哼！你這毒婦竟也來(lái)了冗澈？” 一聲冷哼從身側(cè)響起钦勘，我...
- 序言：老撾萬(wàn)榮一對(duì)情侶失蹤，失蹤者是張志新（化名）和其女友劉穎亚亲，沒(méi)想到半個(gè)月后彻采，有當(dāng)?shù)厝嗽跇?shù)林里發(fā)現(xiàn)了一具尸體，經(jīng)...
- 正文 獨(dú)居荒郊野嶺守林人離奇死亡捌归，尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
- 正文 我和宋清朗相戀三年肛响，在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片惜索。...
- 正文 年R本政府宣布角塑，位于F島的核電站霸奕，受9級(jí)特大地震影響，放射性物質(zhì)發(fā)生泄漏吉拳。R本人自食惡果不足惜，卻給世界環(huán)境...
- 文/蒙蒙 一适揉、第九天 我趴在偏房一處隱蔽的房頂上張望留攒。 院中可真熱鬧煤惩，春花似錦、人聲如沸炼邀。這莊子的主人今日做“春日...
- 文/蒼蘭香墨 我抬頭看了看天上的太陽(yáng)拭宁。三九已至洛退，卻和暖如春，著一層夾襖步出監(jiān)牢的瞬間杰标，已是汗流浹背兵怯。 一陣腳步聲響...
- 正文 我出身青樓，卻偏偏與公主長(zhǎng)得像掸犬，于是被迫代替她去往敵國(guó)和親袜漩。 傳聞我的和親對(duì)象是個(gè)殘疾皇子，可洞房花燭夜當(dāng)晚...
推薦閱讀更多精彩內(nèi)容
- kafka客戶端 Kafka除了提供內(nèi)置Java客戶端外，還提供了二進(jìn)制連接協(xié)議介褥，即向Kafka網(wǎng)絡(luò)端口發(fā)送適當(dāng)?shù)?..
- 此篇開(kāi)始進(jìn)入kafka的另外一側(cè)：消費(fèi)者座掘。kafka中的消費(fèi)者比生產(chǎn)者要復(fù)雜的多，里面涉及到的消費(fèi)組呻顽，偏移量等概念...
- 原文地址 當(dāng)kafka最初被創(chuàng)建的時(shí)候雹顺，它附帶一個(gè)Scala的生產(chǎn)者和消費(fèi)者客戶端。隨著時(shí)間的推移廊遍，我們逐漸意識(shí)到...
- 應(yīng)用程序使用 KafkaConsumer向 Kafka 訂閱主題嬉愧，并從訂閱的主題上接收消息 。 從 Kafka 讀...
