[TOC]
Kafka起初是由 Linkedin公司采用 Scala語言開發(fā)的一個多分區(qū)、多副本且基于 ZooKeeper 協(xié)調(diào)的分布式消息系統(tǒng)，現(xiàn)己被捐獻(xiàn)給 Apache 基金會 。 目前 Kafka 已經(jīng)定位為一個分布式流式 處理平臺，它以高吞吐、可持久化、可水平擴展、支持流數(shù)據(jù)處理等多種特性而被廣泛使用。目前越來越多的開源分布式處理系統(tǒng)如 Cloudera、 Storm、 Spark、 Flink 等都支持與 Kafka 集成 。
Kafka之所以受到越來越多的青睞，與它所“扮演 ”的三大角色是分不開的 :
消息系統(tǒng): Kafka 和傳統(tǒng)的消息系統(tǒng)(也稱作消息中間件〉都具備系統(tǒng)解稿、冗余存 儲、流量削峰、緩沖、異步通信、擴展性、 可恢復(fù)性等功能。與此同時， Kafka 還提 供了大多數(shù)消息系統(tǒng)難以實現(xiàn)的消息 順序性保障及回溯消費 的功能 。
存儲系統(tǒng): Kafka 把消息持久化到磁盤，相比于其他基于內(nèi)存存儲的系統(tǒng)而言，有效 地降低了數(shù)據(jù)丟失的風(fēng)險 。 也正是得益于 Kafka 的消息持久化功能和多副本機制，我 們可以把 Kafka 作為長期的數(shù)據(jù)存儲系統(tǒng)來使用，只需要把對應(yīng)的數(shù)據(jù)保留策略設(shè)置 為“永久”或啟用主題的日志壓縮功能即可 。
流式處理平臺: Kafka 不僅為每個流行的流式處理框架提供了可靠 的數(shù)據(jù)來源，還 提 供了一個完整的流式處理類庫，比如窗口、連接、變換和聚合等各類操作 。
基本概念
一個典型的 Kafka 體系架構(gòu)包括若干 Producer、若干 Broker、若干 Consumer，以及一個 ZooKeeper集群，如圖 1-1 所示。 其中 ZooKeeper是 Kafka用來負(fù)責(zé)集群元數(shù)據(jù)的管理、控制器 的選舉等操作的 。 Producer 將消息發(fā)送到 Broker, Broker 負(fù)責(zé)將收到的消息存儲到磁盤中，而Consumer 負(fù)責(zé)從 Broker 訂閱并消 費消息。
整個 Kafka 體系結(jié)構(gòu)中引入了以下 3 個術(shù)語。
( 1) Producer: 生產(chǎn)者，也就是發(fā)送消息的一方。生產(chǎn)者負(fù)責(zé)創(chuàng)建消息 ， 然后將其投遞到Kafka 中 。
( 2 ) Consumer:消費者，也就是接收消息的 一方。消費者連接到 Kafka 上并接收消息，進(jìn)而進(jìn)行相應(yīng)的業(yè)務(wù)邏輯處理 。
(3) Broker:服務(wù)代理節(jié)點。對于 Kafka 而言， Broker 可以簡單地看作一個獨立的 Kafka 服務(wù)節(jié)點或 Kafka服務(wù)實例。大多數(shù)情況下也可以將 Broker看作一臺 Kafka服務(wù)器，前提是這 臺服務(wù)器上只部署了一個 Kafka 實例。一個或多個 Broker 組成了 一個 Kafka 集群 。一般而言， 我們更習(xí)慣使用首字母小寫的 broker 來表示服務(wù)代理節(jié)點 。
在Kafka中還有兩個特別重要的概念一一主題(Topic)與分區(qū)(Partition)。 Kafka中的消 息以主題為單位進(jìn)行歸類，生產(chǎn)者負(fù)責(zé)將消息發(fā)送到特定的主題(發(fā)送到 Kafka 集群中的每一 條消息都要指定一個主題)，而消費者負(fù)責(zé)訂閱主題并進(jìn)行消費。
主題是一個邏輯上的概念，它還可以細(xì)分為多個分區(qū)，一個分區(qū)只屬于單個主題，很多時 候也會把分區(qū)稱為主題分區(qū)( Topic-Partition)。同一主題下的不同分區(qū)包含的消息是不同的， 分區(qū)在存儲層面可以看作一個可追加的日志( Log)文件，消息在被追加到分區(qū)日志、文件的時 候都會分配一個特定的偏移量(offset)。 offset是消息在分區(qū)中的唯一標(biāo)識， Kafka通過它來保 證消息在分區(qū)內(nèi)的順序性，不過 offset并不跨越分區(qū)，也就是說， Kafka保證的是分區(qū)有序而不 是主題有序。
如圖 1-2 所示，主題中有 4 個分區(qū)，消息被順序追加到每個分區(qū)日志文件的尾部。 Kafka中的分區(qū)可以分布在不同的服務(wù)器 (broker)上，也 就是說，一個主題可以橫跨多個 broker，以 此來提供比單個 broker 更強大的性能 。
每一條消息被發(fā)送到 broker 之前，會根據(jù)分區(qū)規(guī)則選擇存儲到哪個具體的分區(qū) 。 如果分區(qū) 規(guī)則設(shè)定得合理，所有的消息都可以均勻地分配到不同的分區(qū)中 。 如果一個主題只對應(yīng)一個文 件，那么這個文件所在的機器 I/O 將會成為這個主題的性能瓶頸，而分區(qū)解決了這個問題 。 在 創(chuàng)建主題的時候可以通過指定的參數(shù)來設(shè)置分區(qū)的個數(shù)，當(dāng)然也可以在主題創(chuàng)建完成之后去修 改分區(qū)的數(shù)量，通過增加分區(qū)的數(shù)量可以實現(xiàn)水平擴展。
Kafka 為分區(qū)引入了多副本 (Replica) 機制， 通過增加副本數(shù)量可以提升容災(zāi)能力。同一 分區(qū)的不同副本中保存的是相同的消息(在同一時刻，副本之間并非完全一樣)，自1J本之間是 “一主多從”的關(guān)系，其中 leader副本負(fù)責(zé)處理讀寫請求， follower副本只負(fù)責(zé)與 leader副本的 消息同步。副本處于不同的 broker 中 ，當(dāng) leader 副本出現(xiàn)故障時，從 follower 副本中重新選舉 新的 leader副本對外提供服務(wù)。 Kafka通過多副本機制實現(xiàn)了故障的自動轉(zhuǎn)移，當(dāng) Kafka集群中某個 broker 失效時仍然能保證服務(wù)可用 。
如圖1-3所示， Kafka集群中有4個broker，某個主題中有3個分區(qū)，且副本因子(即副本個數(shù)〉也為 3，如此每個分區(qū)便有 l 個 leader副本和 2個 follower副本 。生產(chǎn)者和消費者只與 leader 副本進(jìn)行交互，而 follow巳r副本只負(fù)責(zé)消息的同步，很多時候 follower副本中的消息相對 leader 副本而言會有一定的滯后。
Kafka 消費端也具備一定的容災(zāi)能力。 Consumer 使用拉 (Pull)模式從服務(wù)端拉取消息， 并且保存消費 的具體位置 ， 當(dāng)消費者看機后恢復(fù)上線時可以根據(jù)之前保存的消費位置重新拉取 需要 的消息進(jìn)行消 費 ，這樣就不 會造 成消息丟失 。
分區(qū)中 的所有副本統(tǒng)稱為 AR ( Assigned Replicas) 。 所有與 leader 副本保持 一定程度 同步 的副本(包括 leader 副本在內(nèi)〕組成 ISR On-Sync Replicas ) , ISR 集合是 AR 集合中 的一個子 集 。 消息會先發(fā)送到 lead巳r 副本，然后 follower 副本才能從 leader 副本中拉取消息進(jìn)行同步， 同步期間內(nèi) follower 副本相對于 leader 副本而言會有一定程度的滯后。 前面所說的“一定程度 的同步”是指可忍受的滯后范圍，這個范圍可以通過參數(shù)進(jìn)行配置 。 與 leader 副本同 步滯后過 多的副本(不包 括 leader 副本)組成 OSR ( Out-of-Sync Replicas)，由 此可見， AR=ISR+OSR。 在正常情況下， 所有的 follower副本都應(yīng)該與 leader副本保持一定程度的同步，即 AR=ISR, OSR 集合為空。
leader 副本負(fù) 責(zé)維護(hù)和跟蹤 ISR 集合中所有 follower 副 本 的滯后狀態(tài)， 當(dāng) follower 副本落后 太多或失效時， leader副本會把它從ISR集合中剔除。 如果OSR集合中有follower副本“追上’p 了 leader副本，那么 leader副本會把它從 OSR集合轉(zhuǎn)移至 ISR集合。 默認(rèn)情況下， 當(dāng) leader副 本發(fā)生故障時，只有在 ISR集合中的副本才有資格被選舉為新的 leader， 而在 OSR集合中的副 本則沒有任何機會(不過這個原則也可以通過修改相應(yīng)的參數(shù)配置來改變) 。
ISR 與 HW 和 LEO 也有緊密的關(guān)系 。 HW 是 High Watermark 的縮寫，俗稱高水位，它標(biāo)識 了一個特定的消息偏移量(offset)，消費者只能拉取到這個 offset之前的消息。
如圖 1-4 所示，它代表一個日志文件，這個日志文件中有 9 條消息，第一條消息的 offset (LogStartOffset)為 0
最后一條消息的 offset為 8, offset為 9 的消息用虛線框表示，代表下一條待寫入 的消息 。日志文件的 HW 為 6，表示消費者只能拉取到 offset 在 0 至 5 之間的消息， 而 offset 為 6 的消息對消 費者而言是不可見 的
LEO 是 Log End Offset 的縮寫，它標(biāo)識當(dāng)前日志文件中下一條待寫入消息 的 offset，圖 1-4 中offset為9的位置即為當(dāng)前日志文件的LEO, LEO的大小相當(dāng)于當(dāng)前日志分區(qū)中最后一條消 息的 offset值加 l。分區(qū) ISR集合中的每個副本都會維護(hù)自身的 LEO，而ISR集合中最小的 LEO，即為分區(qū)的 HW ，對消費者而言只能消費 HW 之前的消息 。
很多資料中誤將圖1-4 offset為5看做HW,而把offset為8的位置看作LEO，這顯然是不對的
為了讓讀者更好地理解ISR集合， 以及HW和LEO之間的關(guān)系， 下面通過一個簡單的示 例來進(jìn)行相關(guān)的說明 。 如圖 1-5 所示，假設(shè)某個分區(qū)的 ISR 集合中有 3 個副本，即一個 leader副本和 2 個 follower 副本，此時分區(qū)的 LEO 和 HW 都為 3。消息 3 和消息 4 從生產(chǎn)者發(fā)出之后會被先存入 leader 副本，如圖 1-6 所示 。
在消息寫入leader副本之后， follower 副本會發(fā)送拉取請求來拉取消息3和消息4以進(jìn)行消息同步。
在同步過程中，不同的 follower 副本的同步效率也不盡相同。如圖 1-7 所示， 在某一時刻 follower! 完全跟上了 leader 副本而 follower2 只同步了消息 3，如此 leader 副本的 LEO 為 5, follower! 的 LEO 為 5, follower2 的 LEO 為 4， 那么當(dāng)前分區(qū)的 HW 取最小值 4，此時消費者可以消費到 offset為 0至 3 之間的消息。
寫入消息(情形的如圖 1-8 所示 ，所有的副本都成功 寫入 了消息 3 和消息 4，整個分區(qū)的 HW 和 LEO 都變?yōu)?5，因此消費者可以消費到 offset為 4 的消息了 。
由此可見， Kafka 的復(fù)制機制既不是完全的同步復(fù)制，也不是單純的異步復(fù)制。事實上， 同步復(fù)制要求所有能工作的 folower 副本都復(fù)制完，這條消息才會被確認(rèn)為已成功提交，這種 復(fù)制方式極大地影響了性能。而在異步復(fù)制方式下， follower 副本異步地從 leader 副本中 復(fù)制數(shù) 據(jù)，數(shù)據(jù)只要被 leader 副本寫入就被認(rèn)為已經(jīng)成功提交。在這種情況下，如果 follower 副本都 還沒有復(fù)制完而落后于 leader 副本，突然 leader 副本著機，則會造成數(shù)據(jù)丟失。 Kafka 使用的這 種 ISR 的方式則有效地權(quán)衡了數(shù)據(jù)可靠性和性能之間的關(guān)系。
生產(chǎn)與消費
由 1.1 節(jié)的內(nèi)容可知， 生產(chǎn)者將消息發(fā)送至 Kafka 的主題中， 或者更加確切地說應(yīng)該是主 題的分區(qū)中，而消費者也是通過訂閱主題從而消費消息的。 在演示生產(chǎn)與消費消息之前，需要 創(chuàng)建一個主題作為消息的載體。
Kafka提供了許多實用的腳本工具 ， 存放在$KAFKA HOME 的 bin 目錄下，其中與主題有 關(guān)的就是 kafka-topics.sh 腳本，下面我們用它演示創(chuàng)建一個分區(qū)數(shù)為 4、副本因子為 3 的主題 topic-demo，示例如 下 :
[root@nodel kafka 2.11-2.0.0]# bin/kafka-topics.sh --zookeeper localhost: 2181/kafka --create --topic topic-demo --replication-factor 3 --partitions 4
Created topic "topic- demo” .
其中一zookeeper指定了 Kafka所連接的 ZooKeeper服務(wù)地址，--topic指定了所要創(chuàng) 建主題的名稱， --replication-factor 指定了副本因子， --partitions 指定了分區(qū)個 數(shù)，一create 是創(chuàng)建主題的動作指令。
還可以通過一describe展示主題的更多具體信息， 示例如下:
[root@nodel kafka 2.11-2.0.0]# bin/kafka-topics .sh --zookeeper localhost: 2181/kafka --describe - t opic topic- demo
創(chuàng)建主題 topic-demo 之后我們再來檢測 一 下 Kafka 集群 是否 可以正常地發(fā)送和消費消息 。 $KAFKA HOME/bin 目錄下還提供了兩個腳本 kafka-console-producer.sh 和 kafka-console- consumer. sh，通過控制臺收發(fā)消息。首先我們打開一個 shell終端，通過 kafka-console-consumer.sh 腳本來訂閱 主題 topic-demo，示例如下:
[root@nodel kafka 2.11-2.0.0J# bin/kafka console co口sumer. sh --bootstrap- server localhost:9092 --topic topic- demo
其中一-bootstrap server 指定了連接的 Kafka集群地址，--top工c指定了消費者訂閱 的主題 。 目前主題 topic-demo 尚未有任何消息存入，所以此腳 本還不能消費 任何消 息。
我們再打開一個 shell終端，然后使用 kafka-console-producer.sh腳本發(fā)送一條消息“Hello, Kafka!”至主題 topic-demo，示例如下:
[root@nodel kafka 2.11-2.0.0]# b工n/kafka-console-producer.sh --broker-list localhost: 9092 topic top工c-demo
>hello , Kafka! >
其中 一-broker-list 指定了連接的 Kafka集群地址， --topic 指定了發(fā)送消息時的主題。 示例中的第二行是通過人工鍵入的方式輸入的，按下回車鍵后會跳到第三行，即“>”字符處。 此時原先執(zhí)行 kafka-console-consumer.sh腳本的 shell終端中出現(xiàn)了剛剛輸入的消息、 “Hello, Kafka!”， 示 例如下:
[root@nodel kafka 2.11-2.0.0]# bin/kafka-console-consumer.sh --bootstrap- server localhost:9092 --topic topic- demo
hello world
也可以通過輸入一些其他自定義的消息來熟悉消息的收發(fā)及這兩個腳本的用法 。 不過 這兩個腳本一般用來做一些測試類的工作，在實際應(yīng)用中，不會只是簡單地使用這兩個腳本來 做復(fù)雜的與業(yè)務(wù)邏輯相關(guān)的消息生產(chǎn)與消費的工作，具體的工作還需要通過編程的手段來實施
服務(wù)踹參數(shù)配置
在Kafka 安裝與配置的說明中只是簡單地表述了幾個必要的服務(wù)端參數(shù)而沒有對 其進(jìn)行詳細(xì)的介紹，井且Kafka服務(wù)端參數(shù)(brokerconfigs)也并非只有這幾個。 Kafka服務(wù)端 還有很多參數(shù)配置，涉及使用、調(diào)優(yōu)的各個方面，雖然這些參數(shù)在大多數(shù)情況下不需要更改， 但了解這些參數(shù)，以及在特殊應(yīng)用需求的情況下進(jìn)行有針對性的調(diào)優(yōu)，可以更好地利用 Kafka 為我們工作。下面挑選一些重要的服務(wù)端參數(shù)來做細(xì)致的說明，這些參數(shù)都配置在 $KAFKA_HOME/config/server.properties 文件中。
- zookeeper.connect
該參數(shù)指 明 broker 要連接的 ZooKeeper集群的服務(wù)地址(包含端口號)，沒有默認(rèn)值，且 此參數(shù)為必填工頁。可以配置為 localhost:2181，如果 ZooKeeper集群中有多個節(jié)點，則可以用逗 號將每個節(jié)點隔開，類似于 localhost1 :2181,localhost2 :2181,localhost3: 2181 這種格式。最佳的實踐方式是再加一個 chroot路徑，這樣既可以明確指明該 chroot路徑下的節(jié) 點是為 Kafka 所用的， 也可以 實現(xiàn)多個 Kafka 集群復(fù)用一套 ZooKeeper 集群，這樣可以節(jié)省更 多的硬件資源。包含 chroot 路徑的配置類似 于 localhost1: 2181 , localhost2:2181, localhost3 : 2181/kafka 這種，如果不指定 chroot，那么默認(rèn)使用 ZooKeeper 的根路徑。
- listeners
該參數(shù)指明 broker監(jiān)聽客戶端連接的地址列表，即為客戶端要連接 broker 的入口地址列表， 配置格式為 protocoll : //hostnamel:portl, protocol2://hostname2:port2 ，其 中 protocol 代表協(xié)議類型， Kafka 當(dāng)前支持的協(xié)議類型有 PLAINTEXT、 SSL、 SASL_SSL 等， 如果未開啟安全認(rèn)證，則使用簡單的 PLAINTEXT 即可。 hostname 代表主機名， p。此代表服務(wù) 端口，此參數(shù)的默認(rèn)值為 null。比如此參數(shù)配置為 PLAINTEXT://198.162.0.2:9092，如 果有多個地址，則中間以逗號隔開。如果不指定主機名，則表示綁定默認(rèn)網(wǎng)卡，注意有可能會 綁定到 127.0.0.1，這樣無法對外提供服務(wù)，所以主機名最好不要為 空; 如果主機名是 0.0.0.0, 則表示綁定所有的網(wǎng)卡。與此參數(shù)關(guān)聯(lián)的還有 advertised.listeners， 作用和 listeners 類似，默認(rèn)值也為 null。不過 advertised.listeners 主要用于 IaaS (Infrastructure as a Service)環(huán)境，比如公 有云上的機器通常配備有多塊網(wǎng)卡 ，即包含私網(wǎng)網(wǎng)卡和公網(wǎng)網(wǎng)卡，對于 這種情況而言，可以設(shè)置 advertised.listeners 參數(shù)綁定公網(wǎng) IP 供外部客戶端使用，而 配置 listeners 參數(shù)來綁定私 網(wǎng) IP 地址供 broker 間通信使用 。
broker.id
該參數(shù)用來指定 Kafka 集群中 broker 的唯一標(biāo)識，默認(rèn)值為 1。如果沒有設(shè)置，那么 Kafka會自動生成一個 。這個參數(shù)還和 meta.properties 文件及服務(wù)端參數(shù) broker.id.geηeration. enable 和 reserved.broker.max.id有關(guān)log.dir 和 log.dirs
Kafka 把所有的消息都保存在磁盤上，而這兩個參數(shù)用來配置 Kafka 日志文件存放的根目 錄。一般情況下， log . dir 用來配置單個根目錄，而 log . dirs 用來配置多個根目錄(以逗 號分隔〉，但是 Kafka 井沒有對此做強制性限制，也就是說， log.dir 和 log.dirs 都可以 用來配置單個或多個根目錄 。 log.dirs 的優(yōu)先級比 log.dir 高，但是如果沒有配置 log . dirs ，則會以 log . dir 配置為準(zhǔn)。默認(rèn)情況下只配置了 log . dir 參數(shù)，其默認(rèn)值為 /tmp/kafka-logs。message.max.bytes
該參數(shù)用來指定 broker所能接收消息的最大值，默認(rèn)值為 1000012 (B)，約等于 976.6阻。 如果 Producer 發(fā)送的消息大于這個參數(shù)所設(shè)置的值，那么( Producer 〉就會報出 RecordTooLargeException 的 異常。如果需要修改這個參數(shù)，那么還要考慮 max.request . size(客戶端參數(shù))、 max.message.bytes (topic端參數(shù))等參數(shù)的影響。為了避免修改此參數(shù) 而引起級聯(lián)的影響，建議在修改此參數(shù)之前考慮分拆消息的可行性。
