原文地址：http://blog.csdn.net/honglei915/article/details/37564871

Kafka視頻教程同步首發(fā)，歡迎觀看！

Kafka Producer

消息發(fā)送

producer直接將數(shù)據(jù)發(fā)送到broker的leader(主節(jié)點)，不需要在多個節(jié)點進行分發(fā)。為了幫助producer做到這點，所有的Kafka節(jié)點都可以及時的告知:哪些節(jié)點是活動的，目標topic目標分區(qū)的leader在哪。這樣producer就可以直接將消息發(fā)送到目的地了。

客戶端控制消息將被分發(fā)到哪個分區(qū)。可以通過負載均衡隨機的選擇，或者使用分區(qū)函數(shù)。Kafka允許用戶實現(xiàn)分區(qū)函數(shù)，指定分區(qū)的key，將消息hash到不同的分區(qū)上(當然有需要的話，也可以覆蓋這個分區(qū)函數(shù)自己實現(xiàn)邏輯).比如如果你指定的key是user id，那么同一個用戶發(fā)送的消息都被發(fā)送到同一個分區(qū)上。經過分區(qū)之后，consumer就可以有目的的消費某個分區(qū)的消息。

異步發(fā)送

批量發(fā)送可以很有效的提高發(fā)送效率。Kafka producer的異步發(fā)送模式允許進行批量發(fā)送，先將消息緩存在內存中，然后一次請求批量發(fā)送出去。這個策略可以配置的，比如可以指定緩存的消息達到某個量的時候就發(fā)出去，或者緩存了固定的時間后就發(fā)送出去（比如100條消息就發(fā)送，或者每5秒發(fā)送一次）。這種策略將大大減少服務端的I/O次數(shù)。

既然緩存是在producer端進行的，那么當producer崩潰時，這些消息就會丟失。Kafka0.8.1的異步發(fā)送模式還不支持回調，就不能在發(fā)送出錯時進行處理。Kafka 0.9可能會增加這樣的回調函數(shù)。見Proposed Producer API.

Kafka Consumer

Kafa consumer消費消息時，向broker發(fā)出"fetch"請求去消費特定分區(qū)的消息。consumer指定消息在日志中的偏移量（offset），就可以消費從這個位置開始的消息。customer擁有了offset的控制權，可以向后回滾去重新消費之前的消息，這是很有意義的。 ?

推還是拉？

Kafka最初考慮的問題是，customer應該從brokes拉取消息還是brokers將消息推送到consumer，也就是pull還push。在這方面，Kafka遵循了一種大部分消息系統(tǒng)共同的傳統(tǒng)的設計：producer將消息推送到broker，consumer從broker拉取消息。
一些消息系統(tǒng)比如Scribe和Apache Flume采用了push模式，將消息推送到下游的consumer。這樣做有好處也有壞處：由broker決定消息推送的速率，對于不同消費速率的consumer就不太好處理了。消息系統(tǒng)都致力于讓consumer以最大的速率最快速的消費消息，但不幸的是，push模式下，當broker推送的速率遠大于consumer消費的速率時，consumer恐怕就要崩潰了。最終Kafka還是選取了傳統(tǒng)的pull模式。
Pull模式的另外一個好處是consumer可以自主決定是否批量的從broker拉取數(shù)據(jù)。Push模式必須在不知道下游consumer消費能力和消費策略的情況下決定是立即推送每條消息還是緩存之后批量推送。如果為了避免consumer崩潰而采用較低的推送速率，將可能導致一次只推送較少的消息而造成浪費。Pull模式下，consumer就可以根據(jù)自己的消費能力去決定這些策略。
Pull有個缺點是，如果broker沒有可供消費的消息，將導致consumer不斷在循環(huán)中輪詢，直到新消息到t達。為了避免這點，Kafka有個參數(shù)可以讓consumer阻塞知道新消息到達(當然也可以阻塞知道消息的數(shù)量達到某個特定的量這樣就可以批量發(fā)送)。

消費狀態(tài)跟蹤

對消費消息狀態(tài)的記錄也是很重要的。
大部分消息系統(tǒng)在broker端的維護消息被消費的記錄：一個消息被分發(fā)到consumer后broker就馬上進行標記或者等待customer的通知后進行標記。這樣也可以在消息在消費后立馬就刪除以減少空間占用。
但是這樣會不會有什么問題呢？如果一條消息發(fā)送出去之后就立即被標記為消費過的，一旦consumer處理消息時失敗了（比如程序崩潰）消息就丟失了。為了解決這個問題，很多消息系統(tǒng)提供了另外一個個功能：當消息被發(fā)送出去之后僅僅被標記為已發(fā)送狀態(tài)，當接到consumer已經消費成功的通知后才標記為已被消費的狀態(tài)。這雖然解決了消息丟失的問題，但產生了新問題，首先如果consumer處理消息成功了但是向broker發(fā)送響應時失敗了，這條消息將被消費兩次。第二個問題時，broker必須維護每條消息的狀態(tài)，并且每次都要先鎖住消息然后更改狀態(tài)然后釋放鎖。這樣麻煩又來了，且不說要維護大量的狀態(tài)數(shù)據(jù)，比如如果消息發(fā)送出去但沒有收到消費成功的通知，這條消息將一直處于被鎖定的狀態(tài)，
Kafka采用了不同的策略。Topic被分成了若干分區(qū)，每個分區(qū)在同一時間只被一個consumer消費。這意味著每個分區(qū)被消費的消息在日志中的位置僅僅是一個簡單的整數(shù)：offset。這樣就很容易標記每個分區(qū)消費狀態(tài)就很容易了，僅僅需要一個整數(shù)而已。這樣消費狀態(tài)的跟蹤就很簡單了。
這帶來了另外一個好處：consumer可以把offset調成一個較老的值，去重新消費老的消息。這對傳統(tǒng)的消息系統(tǒng)來說看起來有些不可思議，但確實是非常有用的，誰規(guī)定了一條消息只能被消費一次呢？consumer發(fā)現(xiàn)解析數(shù)據(jù)的程序有bug，在修改bug后再來解析一次消息，看起來是很合理的額呀！

離線處理消息

高級的數(shù)據(jù)持久化允許consumer每個隔一段時間批量的將數(shù)據(jù)加載到線下系統(tǒng)中比如Hadoop或者數(shù)據(jù)倉庫。這種情況下，Hadoop可以將加載任務分拆，拆成每個broker或每個topic或每個分區(qū)一個加載任務。Hadoop具有任務管理功能，當一個任務失敗了就可以重啟而不用擔心數(shù)據(jù)被重新加載，只要從上次加載的位置繼續(xù)加載消息就可以了。

總結

以上是生活随笔為你收集整理的漫游Kafka设计篇之Producer和Consumer的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。