kafka采用Consumer消費(fèi)者Pull主動(dòng)拉取數(shù)據(jù)的方式，當(dāng)Broker無(wú)數(shù)據(jù)時(shí)，消費(fèi)者空轉(zhuǎn) 。Kafka并不刪除已消費(fèi)的消息，各自獨(dú)立的消費(fèi)者可消費(fèi)同一個(gè)Broker分區(qū)數(shù)據(jù) 。
消費(fèi)流程1、消費(fèi)者發(fā)起網(wǎng)絡(luò)消費(fèi)請(qǐng)求
- # 每批次最小抓取設(shè)置（推薦1字節(jié)）
- fetch.min.bytes
- # 每批次最大抓取大小設(shè)置（推薦500ms）
- fetch.max.bytes
- # 未達(dá)到大小的超時(shí)設(shè)置（推薦50M）
- fetch.max.wait.ms
- # 單次拉取最大消息條數(shù)設(shè)置（推薦500條）
- max.poll.records
2.2、攔截器處理（如：匯總統(tǒng)計(jì)記錄）
3、數(shù)據(jù)的后續(xù)處理保存等的消費(fèi)端動(dòng)作 。
offset當(dāng)一個(gè)消費(fèi)者掛掉或重啟后，是否還記得消費(fèi)到的位置了？offset解決了此問(wèn)題 。
對(duì)于每一個(gè)topic，都會(huì)維持一個(gè)分區(qū)日志，分區(qū)中的每一個(gè)記錄都會(huì)分配一個(gè)Id來(lái)表示順序，稱之為offset，offset用來(lái)唯一的標(biāo)識(shí)分區(qū)中每條記錄，并將每次的消費(fèi)位置提交到topic中 。消費(fèi)者恢復(fù)啟動(dòng)后接著按序消費(fèi)數(shù)據(jù) 。
自動(dòng)提交
- # 開(kāi)啟自動(dòng)提交
- enable.auto.commit = true
- # 每次提交間隔（推薦5秒）
- auto.commit.interval.ms = 5000
- # 同步提交，等提交完成才可下一次再消費(fèi)
- .CommitSync
- # 異步提交，可直接進(jìn)行下一個(gè)消費(fèi)，也有可能提交失敗
- .CommitAync
- # 按指定時(shí)間得出offset值
- .offsetsForTimes
- # 按指定offset值繼續(xù)消費(fèi)
- .seek
