上圖所示，consumer訂閱kafka叢集中(一個broker中的一個topic中)的訊息，然後對broker發起一個獲取訊息的請求，請求中攜帶了topic、partition、offset等資訊，接著用pull的方式獲取kafka log中所有可用訊息，並對訊息中的資料進行處理，比如使用spark進行計算，將結果存入DB中。 consumer訂閱訊息時，會連線上任一個可用的broker，並獲取topic中leader partition的元資料metadata資訊，這樣consumer就可以直接與leader partition通訊，獲取訊息。 消費者Consumer客戶端可以用多種語言實現，如Java語言、C語言或者Python語言。這裡採用Java語言實現。 Consumer通過與brokers的TCP連線來獲取資料。未能及時關閉Consumer，將導致TCP連線洩露。 Consumer是非執行緒安全的； offset 在kafka中的每一個partition中 ，kafka會為每一條訊息記錄分配一個數值型的offset。offset值唯一標識了partition中的一條訊息，也表示消費者在分割槽的消費位置。也就是說，一個消費者Consumer的消費position為5，說明已經消費了offset為0,1,2,3,4的訊息，下一個要消費的訊息的offset為5。position對於Consumer來說，有兩層含義： position: Consumer的position等於將要消費記錄的offset；大於Consumer已消費paitition中的訊息的offset的最大值。在Consumer每次呼叫poll(long)方法獲取訊息時，position的值自動增加。 committed position：是最後一個成功儲存的offset，重啟或者錯誤處理時，需要恢復到的offset的值等於committed position。Comsumer可以週期性的自動提交offset，也可以手工呼叫commitSync來或者commitAsync方法提交。commitSync方法將阻塞，直到提交成功或者發生錯誤。commitAsync方法不阻塞，無論成功或者失敗，都將呼叫回撥函式OffsetCommitCallback。 消費者分組和再平衡 kafka使用consumer groups來劃分訊息處理和消費的程序(池)。這些程序可以執行在同一臺機器上，也可以執行在多臺機器上(有利於擴充套件和容錯)。 一個消費者Comsumer只能屬於一個consumer group ,通過subscribe API 可以動態設定topic列表。kafka會將topic中的每一條訊息傳送給consumer group中的一條程序。為了使topic partition分割槽與consumer group中的程序達到平衡，每一個partition只會有consumer group中的一個消費者來消費。例如：如果一個topic有4個分割槽，一個consumer group有2個程序(消費者)，那麼，每一個程序會消費2個分割槽中的訊息。 如果consumer group分組中的一個consumer失敗了，其消費的topic分割槽將會分配給相同分組中的其他消費者；如果consumer group分組中新增了一個consumer，topic分割槽將會從已有消費者上移動到新消費者上。上述過程稱為rebalancing再平衡。當topic中增加一個新分割槽時，會採用相同的過程進行再平衡。 一個consumer group可以看做是一個獨立的邏輯訂閱者，此訂閱者可以包含多個程序。 當一個分組進行再平衡操作時，會通過 ConsumerRebalanceListener類來通知消費者，消費者可以藉此進行應用程式級別的邏輯處理：如狀態清零、手工提交offset等。 一個partition只會被一個消費者消費。 消費者consumer失敗檢測 消費者訂閱一系列topic主題之後，在呼叫poll(long)方法時，將自動歸屬於一個group。poll API來保證consumer的存活狀態。只要持續呼叫poll方法，消費者將一直存在於分組中，持續收到所屬partition中的訊息。底層實現內幕是：poll API會週期性的向server傳送心跳，當停止呼叫poll方法時，將會停止傳送心跳。如果在超過了session失效時間，那麼此消費者會被從當前組中移除，其消費的partition將會被重新分配。這樣做，是為了避免這樣的場景：一個消費者失敗了，仍然持有其partition。 這樣的設計意味著：poll迴圈中，訊息的處理時間要小於心跳的超時時間。如果大於心跳超時時間，消費者將無法提交offset（commitSync()方法會丟擲CommitFailedException 異常）。 消費者提供了兩種配置設定來控制這種行為： 1、session.timeout.ms:通過增加session失效時間，consumer可以有更多的時間來處理從poll(long)獲取的一批記錄。缺點是：延長了server發現consumer失敗的時間，進而導致延遲再平衡時間。但是不包括consumer呼叫close方法的情況，因為此時consumer會發送一個顯式的訊息到server，此時會觸發一個及時的再平衡操作。 2、max.poll.records: poll迴圈中的處理時間與處理的記錄數量成正比，所以要限制一次處理記錄的數量。可以通過此引數設定，預設情況下是沒有限制。 在一些場景下，訊息處理時間是很難預測的，上述兩種配置都不可行。推薦的方式為：將訊息處理邏輯放到一個獨立的執行緒中，這樣consumer可以繼續傳送心跳。需要注意的是，提交的offset不應該在實際位置之前。也就是說，你應該禁用自動提交offset，在訊息處理執行緒中手工提交offset。通常情況下，你需要使用 pause(Collection)方法來停止從partition中獲取新的訊息。消費者api提供了靈活性,覆蓋了各種消費的用例。