本章結(jié)構(gòu)：

1、為什么要使用消息隊(duì)列MQ

2、使用消息隊(duì)列的好處

3、消息隊(duì)列的兩種模式

4、對(duì)Kafka的概述

5、Kafka的特性

6、Kafka的系統(tǒng)架構(gòu)

7、部署Kafka

Kafka 定義

Kafka 是一個(gè)分布式的基于發(fā)布/訂閱模式的消息隊(duì)列（MQ，Message Queue），主要應(yīng)用于大數(shù)據(jù)實(shí)時(shí)處理領(lǐng)域。

一、為什么使用消息隊(duì)列MQ

在高并發(fā)環(huán)境下，同步請(qǐng)求來(lái)不及處理會(huì)發(fā)生堵塞，從而觸發(fā)too many connection錯(cuò)誤，引發(fā)雪崩效應(yīng)。比如大量的請(qǐng)求并發(fā)訪問(wèn)數(shù)據(jù)庫(kù)，導(dǎo)致行鎖表鎖，最后請(qǐng)求線程會(huì)堆積過(guò)多。

我們使用消息隊(duì)列，通過(guò)異步請(qǐng)求，緩解系統(tǒng)壓力，消息隊(duì)列經(jīng)常應(yīng)用于異步處理，流量削峰，應(yīng)用解耦，消息通訊等場(chǎng)景。

當(dāng)前比較常見(jiàn)的 MQ 中間件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等。

二、使用消息隊(duì)列的好處

1、解耦

允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過(guò)程，只要確保它們遵守同樣的接口約束。

2、可恢復(fù)性

系統(tǒng)的一部分組件失效時(shí)，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛了，加入隊(duì)列的消息仍然可以在系統(tǒng)恢復(fù)后被處理。

3、緩沖

有助于控制和優(yōu)化數(shù)據(jù)流結(jié)果系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況。

4、靈活性，峰值處理能力

在訪問(wèn)量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見(jiàn)。
如果為以能處理這類峰值訪問(wèn)為標(biāo)準(zhǔn)來(lái)投入資源隨時(shí)待命無(wú)疑是巨大的浪費(fèi)。
使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問(wèn)壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰。
?

5、異步通信

很多時(shí)候，用戶不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制，允許用戶把一個(gè)消息放入隊(duì)列，但并不立即處理它。

想向隊(duì)列中放入多少消息就放多少，然后在需要的時(shí)候再去處理它們。
?

三、消息隊(duì)列的兩種模式

（1）點(diǎn)對(duì)點(diǎn)模式（一對(duì)一，消費(fèi)者主動(dòng)拉取數(shù)據(jù)，消息收到后消息清除）

消息生產(chǎn)者生產(chǎn)消息發(fā)送到消息隊(duì)列中，然后消息消費(fèi)者從消息隊(duì)列中取出并且消費(fèi)消息。
消息被消費(fèi)以后，消息隊(duì)列中不再有存儲(chǔ)，所以消息消費(fèi)者不可能消費(fèi)到已經(jīng)被消費(fèi)的消息。

消息隊(duì)列支持存在多個(gè)消費(fèi)者，但是對(duì)一個(gè)消息而言，只會(huì)有一個(gè)消費(fèi)者可以消費(fèi)。

（2）發(fā)布/訂閱模式（一對(duì)多，又叫觀察者模式，消費(fèi)者消費(fèi)數(shù)據(jù)之后不會(huì)清除消息）

消息生產(chǎn)者（發(fā)布）將消息發(fā)布到 topic 中，同時(shí)有多個(gè)消息消費(fèi)者（訂閱）消費(fèi)該消息。和點(diǎn)對(duì)點(diǎn)方式不同，發(fā)布到 topic 的消息會(huì)被所有訂閱者消費(fèi)。
發(fā)布/訂閱模式是定義對(duì)象間一種一對(duì)多的依賴關(guān)系，使得每當(dāng)一個(gè)對(duì)象（目對(duì)標(biāo)象）的狀態(tài)發(fā)生改變，則所有依賴于它的對(duì)象（觀察者對(duì)象）都會(huì)得到通知并自動(dòng)更新。

四、對(duì)Kafka的概述

基于 Zookeepe
Kafka 是最初由 Linkedin 公司開(kāi)發(fā)，是一個(gè)分布式、支持分區(qū)的（partition）、多副本的（replicar 協(xié)調(diào)的分布式消息中間件系統(tǒng)，它的最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場(chǎng)景，比如基于 hadoop 的批處理系統(tǒng)、低延遲的實(shí)時(shí)系統(tǒng)、Spark/Flink 流式處理引擎，nginx 訪問(wèn)日志，消息服務(wù)等等，用 scala 語(yǔ)言編寫，
Linkedin 于 2010 年貢獻(xiàn)給了 Apache 基金會(huì)并成為頂級(jí)開(kāi)源項(xiàng)目。

五、Kafka的特性

●高吞吐量、低延遲
Kafka 每秒可以處理幾十萬(wàn)條消息，它的延遲最低只有幾毫秒。每個(gè) topic 可以分多個(gè) Partition，Consumer Group 對(duì) Partition 進(jìn)行消費(fèi)操作，提高負(fù)載均衡能力和消費(fèi)能力。

●可擴(kuò)展性
kafka 集群支持熱擴(kuò)展

●持久性、可靠性
消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失

●容錯(cuò)性
允許集群中節(jié)點(diǎn)失敗（多副本情況下，若副本數(shù)量為 n，則允許 n-1 個(gè)節(jié)點(diǎn)失敗）

●高并發(fā)
支持?jǐn)?shù)千個(gè)客戶端同時(shí)讀寫

六、Kafka的系統(tǒng)架構(gòu)

（1）Broker ? ? 服務(wù)器
一臺(tái) kafka 服務(wù)器就是一個(gè) broker。一個(gè)集群由多個(gè) broker 組成。一個(gè) broker 可以容納多個(gè) topic。

（2）Topic ? 主題
可以理解為一個(gè)隊(duì)列，生產(chǎn)者和消費(fèi)者面向的都是一個(gè) topic。
類似于數(shù)據(jù)庫(kù)的表名或者 ES 的 index
物理上不同 topic 的消息分開(kāi)存儲(chǔ)

（3）Partition ?分區(qū)
為了實(shí)現(xiàn)擴(kuò)展性，一個(gè)非常大的 topic 可以分布到多個(gè) broker（即服務(wù)器）上，一個(gè) topic 可以分割為一個(gè)或多個(gè) partition，每個(gè) partition 是一個(gè)有序的隊(duì)列。Kafka 只保證 partition 內(nèi)的記錄是有序的，而不保證 topic 中不同 partition 的順序。

每個(gè) topic 至少有一個(gè) partition，當(dāng)生產(chǎn)者產(chǎn)生數(shù)據(jù)的時(shí)候，會(huì)根據(jù)分配策略選擇分區(qū)，然后將消息追加到指定的分區(qū)的隊(duì)列末尾。
##Partation 數(shù)據(jù)路由規(guī)則：
1．指定了 patition，則直接使用；
2．未指定 patition 但指定 key（相當(dāng)于消息中某個(gè)屬性），通過(guò)對(duì) key 的 value 進(jìn)行 hash 取模，選出一個(gè) patition；
3．patition 和 key 都未指定，使用輪詢選出一個(gè) patition。

每條消息都會(huì)有一個(gè)自增的編號(hào)，用于標(biāo)識(shí)消息的偏移量，標(biāo)識(shí)順序從 0 開(kāi)始。

每個(gè) partition 中的數(shù)據(jù)使用多個(gè) segment 文件存儲(chǔ)。

如果 topic 有多個(gè) partition，消費(fèi)數(shù)據(jù)時(shí)就不能保證數(shù)據(jù)的順序。嚴(yán)格保證消息的消費(fèi)順序的場(chǎng)景下（例如商品秒殺、 搶紅包），需要將 partition 數(shù)目設(shè)為 1。

●broker 存儲(chǔ) topic 的數(shù)據(jù)。如果某 topic 有 N 個(gè) partition，集群有 N 個(gè) broker，那么每個(gè) broker 存儲(chǔ)該 topic 的一個(gè) partition。
●如果某 topic 有 N 個(gè) partition，集群有 (N+M) 個(gè) broker，那么其中有 N 個(gè) broker 存儲(chǔ) topic 的一個(gè) partition， 剩下的 M 個(gè) broker 不存儲(chǔ)該 topic 的 partition 數(shù)據(jù)。
●如果某 topic 有 N 個(gè) partition，集群中 broker 數(shù)目少于 N 個(gè)，那么一個(gè) broker 存儲(chǔ)該 topic 的一個(gè)或多個(gè) partition。在實(shí)際生產(chǎn)環(huán)境中，盡量避免這種情況的發(fā)生，這種情況容易導(dǎo)致 Kafka 集群數(shù)據(jù)不均衡。

//分區(qū)的原因
●方便在集群中擴(kuò)展，每個(gè)Partition可以通過(guò)調(diào)整以適應(yīng)它所在的機(jī)器，而一個(gè)topic又可以有多個(gè)Partition組成，因此整個(gè)集群就可以適應(yīng)任意大小的數(shù)據(jù)了；
●可以提高并發(fā)，因?yàn)榭梢砸訮artition為單位讀寫了。

（1）Replica
副本，為保證集群中的某個(gè)節(jié)點(diǎn)發(fā)生故障時(shí)，該節(jié)點(diǎn)上的 partition 數(shù)據(jù)不丟失，且 kafka 仍然能夠繼續(xù)工作，kafka 提供了副本機(jī)制，一個(gè) topic 的每個(gè)分區(qū)都有若干個(gè)副本，一個(gè) leader 和若干個(gè) follower。

（2）Leader
每個(gè) partition 有多個(gè)副本，其中有且僅有一個(gè)作為 Leader，Leader 是當(dāng)前負(fù)責(zé)數(shù)據(jù)的讀寫的 partition。

（3）Follower
Follower 跟隨 Leader，所有寫請(qǐng)求都通過(guò) Leader 路由，數(shù)據(jù)變更會(huì)廣播給所有 Follower，Follower 與 Leader 保持?jǐn)?shù)據(jù)同步。Follower 只負(fù)責(zé)備份，不負(fù)責(zé)數(shù)據(jù)的讀寫。
如果 Leader 故障，則從 Follower 中選舉出一個(gè)新的 Leader。
當(dāng) Follower 掛掉、卡住或者同步太慢，Leader 會(huì)把這個(gè) Follower 從 ISR（Leader 維護(hù)的一個(gè)和 Leader 保持同步的 Follower 集合） 列表中刪除，重新創(chuàng)建一個(gè) Follower。

（4） producer
生產(chǎn)者即數(shù)據(jù)的發(fā)布者，該角色將消息 push 發(fā)布到 Kafka 的 topic 中。
broker 接收到生產(chǎn)者發(fā)送的消息后，broker 將該消息追加到當(dāng)前用于追加數(shù)據(jù)的 segment 文件中。
生產(chǎn)者發(fā)送的消息，存儲(chǔ)到一個(gè) partition 中，生產(chǎn)者也可以指定數(shù)據(jù)存儲(chǔ)的 partition。

（5）Consumer
消費(fèi)者可以從 broker 中 pull 拉取數(shù)據(jù)。消費(fèi)者可以消費(fèi)多個(gè) topic 中的數(shù)據(jù)。

（6）Consumer Group（CG）
消費(fèi)者組，由多個(gè) consumer 組成。
所有的消費(fèi)者都屬于某個(gè)消費(fèi)者組，即消費(fèi)者組是邏輯上的一個(gè)訂閱者?？蔀槊總€(gè)消費(fèi)者指定組名，若不指定組名則屬于默認(rèn)的組。
將多個(gè)消費(fèi)者集中到一起去處理某一個(gè) Topic 的數(shù)據(jù)，可以更快的提高數(shù)據(jù)的消費(fèi)能力。
消費(fèi)者組內(nèi)每個(gè)消費(fèi)者負(fù)責(zé)消費(fèi)不同分區(qū)的數(shù)據(jù)，一個(gè)分區(qū)只能由一個(gè)組內(nèi)消費(fèi)者消費(fèi)，防止數(shù)據(jù)被重復(fù)讀取。
消費(fèi)者組之間互不影響。

（7）offset 偏移量
可以唯一的標(biāo)識(shí)一條消息。
偏移量決定讀取數(shù)據(jù)的位置，不會(huì)有線程安全的問(wèn)題，消費(fèi)者通過(guò)偏移量來(lái)決定下次讀取的消息（即消費(fèi)位置）。
消息被消費(fèi)之后，并不被馬上刪除，這樣多個(gè)業(yè)務(wù)就可以重復(fù)使用 Kafka 的消息。
某一個(gè)業(yè)務(wù)也可以通過(guò)修改偏移量達(dá)到重新讀取消息的目的，偏移量由用戶控制。
消息最終是會(huì)還被刪除的，默認(rèn)生命周期為 1 周（7*24小時(shí)）。

（8）Zookeeper
Kafka 通過(guò) Zookeeper 來(lái)存儲(chǔ)集群的 meta 信息。

由于 consumer 在消費(fèi)過(guò)程中可能會(huì)出現(xiàn)斷電宕機(jī)等故障，consumer 恢復(fù)后，需要從故障前的位置的繼續(xù)消費(fèi)，所以 consumer 需要實(shí)時(shí)記錄自己消費(fèi)到了哪個(gè) offset，以便故障恢復(fù)后繼續(xù)消費(fèi)。
Kafka 0.9 版本之前，consumer 默認(rèn)將 offset 保存在 Zookeeper 中；從 0.9 版本開(kāi)始，consumer 默認(rèn)將 offset 保存在 Kafka 一個(gè)內(nèi)置的 topic 中，該 topic 為 __consumer_offsets。

也就是說(shuō)，zookeeper的作用就是，生產(chǎn)者push數(shù)據(jù)到kafka集群，就必須要找到kafka集群的節(jié)點(diǎn)在哪里，這些都是通過(guò)zookeeper去尋找的。消費(fèi)者消費(fèi)哪一條數(shù)據(jù)，也需要zookeeper的支持，從zookeeper獲得offset，offset記錄上一次消費(fèi)的數(shù)據(jù)消費(fèi)到哪里，這樣就可以接著下一條數(shù)據(jù)進(jìn)行消費(fèi)。