Apache Kafka 最早是由 LinkedIn 開源出來的分布式消息系統(tǒng)，現(xiàn)在是 Apache 旗下的一個子項目射赛，并且已經(jīng)成為開源領(lǐng)域應(yīng)用最廣泛的消息系統(tǒng)之一多柑。
Kafka 社區(qū)非常活躍楣责，從 0.9 版本開始竣灌，Kafka 的標(biāo)語已經(jīng)從“一個高吞吐量聂沙，分布式的消息系統(tǒng)”改為"一個分布式流平臺"。
Kafka 和傳統(tǒng)的消息系統(tǒng)不同在于：
Kafka是一個分布式系統(tǒng)初嘹，易于向外擴展及汉。
它同時為發(fā)布和訂閱提供高吞吐量。
它支持多訂閱者屯烦，當(dāng)失敗時能自動平衡消費者坷随。
消息的持久化。
Kafka 和其他消息隊列的對比：
入門實例
生產(chǎn)者
代碼如下：
importjava.util.Properties;
importorg.apache.kafka.clients.producer.KafkaProducer;
importorg.apache.kafka.clients.producer.ProducerRecord;
publicclassUserKafkaProducerextendsThread
{
privatefinalKafkaProducer?producer;
privatefinalString?topic;
privatefinalProperties?props?=newProperties();
publicUserKafkaProducer(String?topic)
{
props.put("metadata.broker.list","localhost:9092");
props.put("bootstrap.servers","master2:6667");
props.put("retries",0);
props.put("batch.size",16384);
props.put("linger.ms",1);
props.put("buffer.memory",33554432);
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
producer?=newKafkaProducer(props);
this.topic?=?topic;
}
@Override
publicvoidrun(){
intmessageNo?=1;
while(true)
{
String?messageStr?=newString("Message_"+?messageNo);
System.out.println("Send:"+?messageStr);
//返回的是Future<RecordMetadata>,異步發(fā)送
producer.send(newProducerRecord(topic,?messageStr));
messageNo++;
try{
sleep(3000);
}catch(InterruptedException?e)?{
e.printStackTrace();
}
}
}
}
消費者
代碼如下：
Properties?props?=newProperties();
/*?定義kakfa?服務(wù)的地址驻龟，不需要將所有broker指定上?*/
props.put("bootstrap.servers","localhost:9092");
/*?制定consumer?group?*/
props.put("group.id","test");
/*?是否自動確認(rèn)offset?*/
props.put("enable.auto.commit","true");
/*?自動確認(rèn)offset的時間間隔?*/
props.put("auto.commit.interval.ms","1000");
props.put("session.timeout.ms","30000");
/*?key的序列化類?*/
props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
/*?value的序列化類?*/
props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
/*?定義consumer?*/
KafkaConsumer?consumer?=newKafkaConsumer<>(props);
/*?消費者訂閱的topic,?可同時訂閱多個?*/
consumer.subscribe(Arrays.asList("foo","bar"));
/*?讀取數(shù)據(jù)温眉，讀取超時時間為100ms?*/
while(true)?{
ConsumerRecords?records?=?consumer.poll(100);
for(ConsumerRecord?record?:?records)
System.out.printf("offset?=?%d,?key?=?%s,?value?=?%s",?record.offset(),?record.key(),?record.value());
}
Kafka 架構(gòu)原理
對于 Kafka 的架構(gòu)原理，我們先提出如下幾個問題：
Kafka 的 topic 和分區(qū)內(nèi)部是如何存儲的翁狐，有什么特點类溢？
與傳統(tǒng)的消息系統(tǒng)相比，Kafka 的消費模型有什么優(yōu)點?
Kafka 如何實現(xiàn)分布式的數(shù)據(jù)存儲與數(shù)據(jù)讀取?
Kafka 架構(gòu)圖
Kafka 名詞解釋
在一套 Kafka 架構(gòu)中有多個 Producer露懒，多個 Broker豌骏，多個 Consumer，每個 Producer 可以對應(yīng)多個 Topic隐锭，每個 Consumer 只能對應(yīng)一個 Consumer Group窃躲。
整個 Kafka 架構(gòu)對應(yīng)一個 ZK 集群，通過 ZK 管理集群配置钦睡，選舉 Leader蒂窒，以及在 Consumer Group 發(fā)生變化時進行 Rebalance。
Topic 和 Partition
在 Kafka 中的每一條消息都有一個 Topic荞怒。一般來說在我們應(yīng)用中產(chǎn)生不同類型的數(shù)據(jù)洒琢，都可以設(shè)置不同的主題。
一個主題一般會有多個消息的訂閱者褐桌，當(dāng)生產(chǎn)者發(fā)布消息到某個主題時衰抑，訂閱了這個主題的消費者都可以接收到生產(chǎn)者寫入的新消息。
Kafka 為每個主題維護了分布式的分區(qū)(Partition)日志文件荧嵌，每個 Partition 在 Kafka 存儲層面是 Append Log呛踊。
任何發(fā)布到此 Partition 的消息都會被追加到 Log 文件的尾部，在分區(qū)中的每條消息都會按照時間順序分配到一個單調(diào)遞增的順序編號啦撮，也就是我們的 Offset谭网。Offset 是一個 Long 型的數(shù)字。
我們通過這個 Offset 可以確定一條在該 Partition 下的唯一消息赃春。在 Partition 下面是保證了有序性愉择，但是在 Topic 下面沒有保證有序性。
在上圖中我們的生產(chǎn)者會決定發(fā)送到哪個 Partition：
如果沒有 Key 值則進行輪詢發(fā)送。
如果有 Key 值锥涕，對 Key 值進行 Hash衷戈，然后對分區(qū)數(shù)量取余，保證了同一個 Key 值的會被路由到同一個分區(qū)层坠；如果想隊列的強順序一致性殖妇，可以讓所有的消息都設(shè)置為同一個 Key。
消費模型
消息由生產(chǎn)者發(fā)送到 Kafka 集群后窿春，會被消費者消費。一般來說我們的消費模型有兩種：
推送模型(Push)
拉取模型(Pull)
基于推送模型的消息系統(tǒng)采盒，由消息代理記錄消費狀態(tài)旧乞。消息代理將消息推送到消費者后，標(biāo)記這條消息為已經(jīng)被消費磅氨，但是這種方式無法很好地保證消費的處理語義尺栖。
比如當(dāng)我們已經(jīng)把消息發(fā)送給消費者之后，由于消費進程掛掉或者由于網(wǎng)絡(luò)原因沒有收到這條消息烦租，如果我們在消費代理將其標(biāo)記為已消費延赌，這個消息就永久丟失了。
如果我們利用生產(chǎn)者收到消息后回復(fù)這種方法叉橱，消息代理需要記錄消費狀態(tài)挫以，這種不可取。
如果采用 Push窃祝，消息消費的速率就完全由消費代理控制掐松，一旦消費者發(fā)生阻塞，就會出現(xiàn)問題粪小。
Kafka 采取拉取模型(Poll)大磺，由自己控制消費速度，以及消費的進度探膊，消費者可以按照任意的偏移量進行消費杠愧。
比如消費者可以消費已經(jīng)消費過的消息進行重新處理，或者消費最近的消息等等逞壁。
網(wǎng)絡(luò)模型
Kafka Client：單線程 Selector
單線程模式適用于并發(fā)鏈接數(shù)小流济，邏輯簡單，數(shù)據(jù)量小的情況腌闯。在 Kafka 中袭灯，Consumer 和 Producer 都是使用的上面的單線程模式。
這種模式不適合 Kafka 的服務(wù)端绑嘹，在服務(wù)端中請求處理過程比較復(fù)雜稽荧，會造成線程阻塞，一旦出現(xiàn)后續(xù)請求就會無法處理，會造成大量請求超時姨丈，引起雪崩畅卓。而在服務(wù)器中應(yīng)該充分利用多線程來處理執(zhí)行邏輯。
Kafka Server：多線程 Selector
在 Kafka 服務(wù)端采用的是多線程的 Selector 模型蟋恬，Acceptor 運行在一個單獨的線程中翁潘，對于讀取操作的線程池中的線程都會在 Selector 注冊 Read 事件，負(fù)責(zé)服務(wù)端讀取請求的邏輯歼争。
成功讀取后拜马，將請求放入 Message Queue共享隊列中。然后在寫線程池中沐绒，取出這個請求俩莽，對其進行邏輯處理。
這樣乔遮，即使某個請求線程阻塞了扮超，還有后續(xù)的線程從消息隊列中獲取請求并進行處理，在寫線程中處理完邏輯處理蹋肮，由于注冊了 OP_WIRTE 事件出刷，所以還需要對其發(fā)送響應(yīng)。
高可靠分布式存儲模型
在 Kafka 中保證高可靠模型依靠的是副本機制坯辩，有了副本機制之后馁龟，就算機器宕機也不會發(fā)生數(shù)據(jù)丟失。
高性能的日志存儲
Kafka 一個 Topic 下面的所有消息都是以 Partition 的方式分布式的存儲在多個節(jié)點上漆魔。
同時在 Kafka 的機器上屁柏，每個 Partition 其實都會對應(yīng)一個日志目錄，在目錄下面會對應(yīng)多個日志分段(LogSegment)有送。
LogSegment 文件由兩部分組成淌喻，分別為“.index”文件和“.log”文件，分別表示為 Segment 索引文件和數(shù)據(jù)文件雀摘。
這兩個文件的命令規(guī)則為：Partition 全局的第一個 Segment 從 0 開始裸删，后續(xù)每個 Segment 文件名為上一個 Segment 文件最后一條消息的 Offset 值，數(shù)值大小為 64 位阵赠，20 位數(shù)字字符長度涯塔，沒有數(shù)字用 0 填充。
如下清蚀，假設(shè)有 1000 條消息匕荸，每個 LogSegment 大小為 100，下面展現(xiàn)了 900-1000 的索引和 Log：
由于 Kafka 消息數(shù)據(jù)太大枷邪，如果全部建立索引榛搔，既占了空間又增加了耗時，所以 Kafka 選擇了稀疏索引的方式，這樣索引可以直接進入內(nèi)存践惑，加快偏查詢速度腹泌。
簡單介紹一下如何讀取數(shù)據(jù)，如果我們要讀取第 911 條數(shù)據(jù)首先第一步尔觉，找到它是屬于哪一段的凉袱。
根據(jù)二分法查找到它屬于的文件，找到 0000900.index 和 00000900.log 之后侦铜，然后去 index 中去查找 (911-900) = 11 這個索引或者小于 11 最近的索引专甩。
在這里通過二分法我們找到了索引是 [10,1367]，然后我們通過這條索引的物理位置 1367钉稍，開始往后找涤躲，直到找到 911 條數(shù)據(jù)。
上面講的是如果要找某個 Offset 的流程嫁盲，但是我們大多數(shù)時候并不需要查找某個 Offset篓叶，只需要按照順序讀即可烈掠。
而在順序讀中羞秤，操作系統(tǒng)會在內(nèi)存和磁盤之間添加 Page Cache，也就是我們平常見到的預(yù)讀操作左敌，所以我們的順序讀操作時速度很快瘾蛋。
但是 Kafka 有個問題，如果分區(qū)過多矫限，那么日志分段也會很多哺哼，寫的時候由于是批量寫，其實就會變成隨機寫了叼风，隨機 I/O 這個時候?qū)π阅苡绊懞艽笕《Ｋ砸话銇碚f Kafka 不能有太多的 Partition。
針對這一點无宿，RocketMQ 把所有的日志都寫在一個文件里面茵汰，就能變成順序?qū)懀ㄟ^一定優(yōu)化孽鸡，讀也能接近于順序讀蹂午。
大家可以思考一下：
為什么需要分區(qū)，也就是說主題只有一個分區(qū)彬碱，難道不行嗎豆胸？
日志為什么需要分段？
副本機制
Kafka 的副本機制是多個服務(wù)端節(jié)點對其他節(jié)點的主題分區(qū)的日志進行復(fù)制巷疼。
當(dāng)集群中的某個節(jié)點出現(xiàn)故障晚胡，訪問故障節(jié)點的請求會被轉(zhuǎn)移到其他正常節(jié)點(這一過程通常叫 Reblance)。
Kafka 每個主題的每個分區(qū)都有一個主副本以及 0 個或者多個副本，副本保持和主副本的數(shù)據(jù)同步搬泥，當(dāng)主副本出故障時就會被替代桑寨。
在 Kafka 中并不是所有的副本都能被拿來替代主副本，所以在 Kafka 的 Leader 節(jié)點中維護著一個 ISR(In Sync Replicas)集合忿檩。
翻譯過來也叫正在同步中集合尉尾，在這個集合中的需要滿足兩個條件：
節(jié)點必須和 ZK 保持連接。
在同步的過程中這個副本不能落后主副本太多燥透。
另外還有個 AR(Assigned Replicas)用來標(biāo)識副本的全集沙咏，OSR 用來表示由于落后被剔除的副本集合。
所以公式如下：ISR = Leader + 沒有落后太多的副本班套；AR = OSR+ ISR肢藐。
這里先要說下兩個名詞：HW(高水位)是 Consumer 能夠看到的此 Partition 的位置，LEO 是每個 Partition 的 Log 最后一條 Message 的位置吱韭。
HW 能保證 Leader 所在的 Broker 失效吆豹，該消息仍然可以從新選舉的 Leader 中獲取，不會造成消息丟失理盆。
當(dāng) Producer 向 Leader 發(fā)送數(shù)據(jù)時痘煤，可以通過 request.required.acks 參數(shù)來設(shè)置數(shù)據(jù)可靠性的級別：
1（默認(rèn)）：這意味著 Producer 在 ISR 中的 Leader 已成功收到的數(shù)據(jù)并得到確認(rèn)后發(fā)送下一條 Message。如果 Leader 宕機了猿规，則會丟失數(shù)據(jù)衷快。
0：這意味著 Producer 無需等待來自 Broker 的確認(rèn)而繼續(xù)發(fā)送下一批消息。這種情況下數(shù)據(jù)傳輸效率最高姨俩，但是數(shù)據(jù)可靠性卻是最低的蘸拔。
-1：Producer 需要等待 ISR 中的所有 Follower 都確認(rèn)接收到數(shù)據(jù)后才算一次發(fā)送完成，可靠性最高环葵。
但是這樣也不能保證數(shù)據(jù)不丟失调窍，比如當(dāng) ISR 中只有 Leader 時(其他節(jié)點都和 ZK 斷開連接，或者都沒追上)张遭，這樣就變成了 acks = 1 的情況邓萨。
高可用模型及冪等
?在分布式系統(tǒng)中一般有三種處理語義：
at-least-once
至少一次，有可能會有多次帝璧。如果 Producer 收到來自 Ack?的確認(rèn)先誉，則表示該消息已經(jīng)寫入到 Kafka 了，此時剛好是一次的烁，也就是我們后面的 Exactly-once褐耳。
但是如果 Producer 超時或收到錯誤，并且 request.required.acks 配置的不是 -1渴庆，則會重試發(fā)送消息铃芦，客戶端會認(rèn)為該消息未寫入 Kafka雅镊。
如果 Broker 在發(fā)送 Ack 之前失敗，但在消息成功寫入 Kafka 之后刃滓，這一次重試將會導(dǎo)致我們的消息會被寫入兩次仁烹。
所以消息就不止一次地傳遞給最終 Consumer，如果 Consumer 處理邏輯沒有保證冪等的話就會得到不正確的結(jié)果咧虎。
在這種語義中會出現(xiàn)亂序卓缰，也就是當(dāng)?shù)谝淮?Ack 失敗準(zhǔn)備重試的時候，但是第二消息已經(jīng)發(fā)送過去了砰诵，這個時候會出現(xiàn)單分區(qū)中亂序的現(xiàn)象征唬。
我們需要設(shè)置 Prouducer 的參數(shù) max.in.flight.requests.per.connection，flight.requests 是 Producer 端用來保存發(fā)送請求且沒有響應(yīng)的隊列茁彭，保證 Produce r端未響應(yīng)的請求個數(shù)為 1总寒。
at-most-once
如果在 Ack 超時或返回錯誤時 Producer 不重試，也就是我們講 request.required.acks = -1理肺，則該消息可能最終沒有寫入 Kafka摄闸，所以 Consumer 不會接收消息。
exactly-once
剛好一次妹萨，即使 Producer 重試發(fā)送消息年枕，消息也會保證最多一次地傳遞給 Consumer。該語義是最理想的眠副，也是最難實現(xiàn)的画切。
在 0.10 之前并不能保證 exactly-once竣稽，需要使用 Consumer 自帶的冪等性保證囱怕。0.11.0 使用事務(wù)保證了。
如何實現(xiàn) exactly-once
要實現(xiàn) exactly-once 在 Kafka 0.11.0 中有兩個官方策略：
單 Producer 單 Topic
每個 Producer 在初始化的時候都會被分配一個唯一的 PID毫别，對于每個唯一的 PID娃弓，Producer 向指定的 Topic 中某個特定的 Partition 發(fā)送的消息都會攜帶一個從 0 單調(diào)遞增的 Sequence Number。
在我們的 Broker 端也會維護一個維度為岛宦，每次提交一次消息的時候都會對齊進行校驗：
如果消息序號比 Broker 維護的序號大一以上台丛，說明中間有數(shù)據(jù)尚未寫入，也即亂序砾肺，此時 Broker 拒絕該消息挽霉，Producer 拋出 InvalidSequenceNumber。
如果消息序號小于等于 Broker 維護的序號变汪，說明該消息已被保存侠坎，即為重復(fù)消息，Broker 直接丟棄該消息裙盾，Producer 拋出 DuplicateSequenceNumber实胸。
如果消息序號剛好大一他嫡，就證明是合法的。
上面所說的解決了兩個問題：
當(dāng) Prouducer 發(fā)送了一條消息之后失敗庐完，Broker 并沒有保存钢属，但是第二條消息卻發(fā)送成功，造成了數(shù)據(jù)的亂序门躯。
當(dāng) Producer 發(fā)送了一條消息之后淆党，Broker 保存成功，Ack 回傳失敗讶凉，Producer 再次投遞重復(fù)的消息宁否。
上面所說的都是在同一個 PID 下面，意味著必須保證在單個 Producer 中的同一個 Seesion 內(nèi)缀遍，如果 Producer 掛了慕匠，被分配了新的 PID，這樣就無法保證了域醇，所以 Kafka 中又有事務(wù)機制去保證台谊。
事務(wù)
在 Kafka 中事務(wù)的作用是：
實現(xiàn) exactly-once 語義。
保證操作的原子性譬挚，要么全部成功锅铅，要么全部失敗。
有狀態(tài)的操作的恢復(fù)减宣。
事務(wù)可以保證就算跨多個盐须，在本次事務(wù)中的對消費隊列的操作都當(dāng)成原子性，要么全部成功漆腌，要么全部失敗贼邓。
并且，有狀態(tài)的應(yīng)用也可以保證重啟后從斷點處繼續(xù)處理闷尿，也即事務(wù)恢復(fù)塑径。
在 Kafka 的事務(wù)中，應(yīng)用程序必須提供一個唯一的事務(wù) ID填具，即 Transaction ID统舀，并且宕機重啟之后，也不會發(fā)生改變劳景。
Transactin ID 與 PID 可能一一對應(yīng)誉简，區(qū)別在于 Transaction ID 由用戶提供，而 PID 是內(nèi)部的實現(xiàn)對用戶透明盟广。
為了 Producer 重啟之后闷串，舊的 Producer 具有相同的 Transaction ID 失效，每次 Producer 通過 Transaction ID 拿到 PID 的同時衡蚂，還會獲取一個單調(diào)遞增的 Epoch窿克。
由于舊的 Producer 的 Epoch 比新 Producer 的 Epoch 小骏庸，Kafka 可以很容易識別出該 Producer 是老的，Producer 并拒絕其請求年叮。
為了實現(xiàn)這一點具被，Kafka 0.11.0.0 引入了一個服務(wù)器端的模塊，名為 Transaction Coordinator只损，用于管理 Producer 發(fā)送的消息的事務(wù)性一姿。
該 Transaction Coordinator 維護 Transaction Log，該 Log 存于一個內(nèi)部的 Topic 內(nèi)跃惫。
由于 Topic 數(shù)據(jù)具有持久性叮叹，因此事務(wù)的狀態(tài)也具有持久性。Producer 并不直接讀寫 Transaction Log爆存，它與 Transaction Coordinator 通信蛉顽，然后由 Transaction Coordinator 將該事務(wù)的狀態(tài)插入相應(yīng)的 Transaction Log。
Transaction Log 的設(shè)計與 Offset Log 用于保存 Consumer 的 Offset 類似先较。
最后
關(guān)于消息隊列或者 Kafka 的一些常見的面試題携冤，通過上面的文章可以提煉出以下幾個比較經(jīng)典的問題，大部分問題都可以從上面總結(jié)后找到答案：
為什么使用消息隊列闲勺？消息隊列的作用是什么曾棕？
Kafka 的 Topic 和分區(qū)內(nèi)部是如何存儲的，有什么特點菜循？
與傳統(tǒng)的消息系統(tǒng)相比翘地，Kafka 的消費模型有什么優(yōu)點?
Kafka 如何實現(xiàn)分布式的數(shù)據(jù)存儲與數(shù)據(jù)讀取?
Kafka 為什么比 RocketMQ 支持的單機 Partition 要少?
為什么需要分區(qū)，也就是說主題只有一個分區(qū)癌幕，難道不行嗎衙耕？
日志為什么需要分段？
Kafka 是依靠什么機制保持高可靠序芦，高可用臭杰？
消息隊列如何保證消息冪等粤咪？
讓你自己設(shè)計個消息隊列谚中，你會怎么設(shè)計，會考慮哪些方面寥枝？
歡迎工作一到五年的Java工程師朋友們加入Java程序員開發(fā)： 854393687
群內(nèi)提供免費的Java架構(gòu)學(xué)習(xí)資料（里面有高可用宪塔、高并發(fā)、高性能及分布式囊拜、Jvm性能調(diào)優(yōu)某筐、Spring源碼，MyBatis冠跷，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點的架構(gòu)資料）合理利用自己每一分每一秒的時間來學(xué)習(xí)提升自己南誊，不要再用"沒有時間“來掩飾自己思想上的懶惰身诺！趁年輕，使勁拼抄囚，給未來的自己一個交代霉赡！
