前言

上一篇 Kafka 的文章 --- 插曲：大白話帶你認識Kafka 中我們應該已經瞭解了一些關於基礎角色和叢集架構相關的問題，這時候我們應該很想了解一下如何構建生產中的Kafka叢集或者一些相關的運維工具，所以就應運而生了下文，配圖基本沒有，

一、Kafka的生產叢集部署

1.1 方案背景

假設每天叢集需要承載10億資料。一天24小時，晚上12點到凌晨8點幾乎沒多少資料。

使用二八法則估計，也就是80%的資料（8億）會在16個小時湧入，而且8億的80%的資料（6.4億）會在這16個小時的20%時間（3小時）湧入。

QPS計算公式：640000000 ÷ (3x60x60) = 60000，也就是說高峰期的時候Kafka叢集要扛住每秒6萬的併發。

磁碟空間計算，每天10億資料，每條50kb，也就是46T的資料。儲存2個副本（在上一篇中也提到過其實兩個副本會比較好，因為follower需要去leader那裡同步資料，同步資料的過程需要耗費網路，而且需要磁碟空間，但是這個需要根據實際情況考慮），46 * 2 = 92T，保留最近3天的資料。故需要 92 * 3 = 276T

1.2 QPS方面

部署Kafka，Hadoop，MySQL···等核心分散式系統，一般建議直接採用物理機，拋棄使用一些低配置的虛擬機器器的想法。高併發這個東西，不可能是說，你需要支撐6萬QPS，你的叢集就剛好把這6萬併發卡的死死的。加入某一天出一些活動讓資料量瘋狂上漲，那整個叢集就會垮掉。

但是，假如說你只要支撐6w QPS，單臺物理機本身就能扛住4~5萬的併發。所以這時2臺物理機絕對絕對夠了。但是這裡有一個問題，我們通常是建議，公司預算充足，儘量是讓高峰QPS控制在叢集能承載的總QPS的30%左右（也就是叢集的處理能力是高峰期的3~4倍這個樣子），所以我們搭建的kafka叢集能承載的總QPS為20萬~30萬才是安全的。所以大體上來說，需要5~7臺物理機來部署，基本上就很安全了，每臺物理機要求吞吐量在每秒4~5萬條資料就可以了，物理機的配置和效能也不需要特別高。

1.3 磁碟方面

1.3.1 磁碟數量

需要5臺物理機的情況，需要儲存276T的資料，平均下來差不多一臺56T的資料。這個具體看磁碟數和盤的大小

1.3.2 SAS還是SSD

現在我們需要考慮一個問題：是需要SSD固態硬碟，還是普通機械硬碟？

SSD就是固態硬碟，比機械硬碟要快，那麼到底是快在哪裡呢？其實SSD的快主要是快在磁碟隨機讀寫，就要對磁碟上的隨機位置來讀寫的時候，SSD比機械硬碟要快。比如說MySQL這種就應該使用SSD了（MySQL需要隨機讀寫）。比如說我們在規劃和部署線上系統的MySQL叢集的時候，一般來說必須用SSD，效能可以提高很多，這樣MySQL可以承載的併發請求量也會高很多，而且SQL語句執行的效能也會提高很多。

**因為寫磁碟的時候kafka是順序寫的。機械硬碟順序寫的效能機會跟記憶體讀寫的效能是差不多的，所以對於Kafka叢集來說其實使用機械硬碟就可以了。**如果是需要自己創業或者是在公司成本不足的情況下，經費是能夠縮減就儘量縮減的。

1.4 記憶體角度

JVM非常怕出現full gc的情況。kafka自身的jvm是用不了過多堆記憶體的，因為kafka設計就是規避掉用jvm物件來儲存資料，避免頻繁full gc導致的問題，所以一般kafka自身的jvm堆記憶體，分配個10G左右就夠了，剩下的記憶體全部留給os cache。

那伺服器需要多少記憶體呢。我們估算一下，大概有100個topic，所以要保證有100個topic的leader partition的資料在作業系統的記憶體裡。100個topic，一個topic有5個partition。那麼總共會有500個partition。每個partition的大小是1G（在上一篇中的日誌分段儲存中規定了.log檔案不能超過1個G），我們有2個副本，也就是說要把100個topic的leader partition資料都駐留在記憶體裡需要1000G的記憶體。

我們現在有5臺伺服器，所以平均下來每天伺服器需要200G的記憶體，但是其實partition的資料我們沒必要所有的都要駐留在記憶體裡面，只需要25%的資料在記憶體就行，200G * 0.25 = 50G就可以了（因為在叢集中的生產者和消費者幾乎也算是實時的，基本不會出現訊息積壓太多的情況）。所以一共需要60G（附帶上剛剛的10G Kafka服務）的記憶體，故我們可以挑選64G記憶體的伺服器也行，大不了partition的資料再少一點在記憶體，當然如果能夠提供128G記憶體那就更好。

1.5 CPU core

CPU規劃，主要是看你的這個程式裡會有多少個執行緒，執行緒主要是依託多核CPU來執行的，如果你的執行緒特別多，但是CPU核很少，就會導致你的CPU負載很高，會導致整體工作執行緒執行的效率不太高，上一篇的Kafka的網路設計中講過Kafka的Broker的模型。acceptor執行緒負責去接入客戶端的連線請求，但是他接入了之後其實就會把連線分配給多個processor，預設是3個，但是一般生產環境建議大家還是多加幾個，整體可以提升kafka的吞吐量比如說你可以增加到6個，或者是9個。另外就是負責處理請求的執行緒，是一個執行緒池，預設是8個執行緒，在生產叢集裡，建議大家可以把這塊的執行緒數量稍微多加個2倍~3倍，其實都正常，比如說搞個16個工作執行緒，24個工作執行緒。

後臺會有很多的其他的一些執行緒，比如說定期清理7天前資料的執行緒，Controller負責感知和管控整個叢集的執行緒，副本同步拉取資料的執行緒，這樣算下來每個broker起碼會有上百個執行緒。根據經驗4個cpu core，一般來說幾十個執行緒，在高峰期CPU幾乎都快打滿了。8個cpu core，也就能夠比較寬裕的支撐幾十個執行緒繁忙的工作。所以Kafka的伺服器一般是建議16核，基本上可以hold住一兩百執行緒的工作。當然如果可以給到32 cpu core那就最好不過了。

1.6 網路卡

現在的網基本就是千兆網路卡（1GB / s），還有萬兆網路卡（10GB / s）。kafka叢集之間，broker和broker之間是會做資料同步的，因為leader要同步資料到follower上去，他們是在不同的broker機器上的，broker機器之間會進行頻繁的資料同步，傳輸大量的資料。那每秒兩臺broker機器之間大概會傳輸多大的資料量？

高峰期每秒大概會湧入6萬條資料，約每天處理10000個請求，每個請求50kb，故每秒約進來488M資料，我們還有副本同步資料，故高峰期的時候需要488M * 2 = 976M/s的網路頻寬，所以在高峰期的時候，使用千兆頻寬，網路還是非常有壓力的。

綜上描述

10億資料，6w/s的吞吐量，276T的資料，5臺物理機
硬碟：11（SAS） * 7T，7200轉
記憶體：64GB/128GB，JVM分配10G，剩餘的給os cache
CPU：16核/32核
網路：千兆網路卡，萬兆更好
複製程式碼

二、Kafka的叢集搭建

進到Kafka的config資料夾下，會發現有很多很多的配置檔案，可是都不需要你來修改，你僅僅需要點開一個叫作server.properties的檔案就夠了。

【broker.id】
每個broker都必須自己設定的一個唯一id，可以在0~255之間

【log.dirs】
這個極為重要，kafka的所有資料就是寫入這個目錄下的磁碟檔案中的，如果說機器上有多塊物理硬碟，那麼可以把多個目錄掛載到不同的物理硬碟上，然後這裡可以設定多個目錄，這樣kafka可以資料分散到多塊物理硬碟，多個硬碟的磁頭可以並行寫，這樣可以提升吞吐量。ps：多個目錄用英文逗號分隔

【zookeeper.connect】
連線kafka底層的zookeeper叢集的

【Listeners】
broker監聽客戶端發起請求的埠號，預設是9092

【num.network.threads】預設值為3
【num.io.threads】預設值為8
細心的朋友們應該已經發現了，這就是上一篇我們在網路架構上提到的processor和處理執行緒池的執行緒數目。
所以說掌握Kafka網路架構顯得尤為重要。
現在你看到這兩個引數，就知道這就是Kafka叢集效能的關鍵引數了

【unclean.leader.election.enable】
預設是false，意思就是隻能選舉ISR列表裡的follower成為新的leader，1.0版本後才設為false，之前都是true，允許非ISR列表的follower選舉為新的leader

【delete.topic.enable】
預設true，允許刪除topic

【log.retention.hours】
可以設定一下，要保留資料多少個小時，這個就是底層的磁碟檔案，預設保留7天的資料，根據自己的需求來就行了

【min.insync.replicas】
acks=-1（一條資料必須寫入ISR裡所有副本才算成功），你寫一條資料只要寫入leader就算成功了，不需要等待同步到follower才算寫成功。但是此時如果一個follower宕機了，你寫一條資料到leader之後，leader也宕機，會導致資料的丟失。
複製程式碼

因為實際的叢集搭建說真的沒有太大難度，所以搭建的過程就不詳細展開了，網上應該很多相關資料

三、Kafka的簡單叢集操作

在操作Kafka叢集的時候，不同的Kafka版本命令的寫法是不一樣的，所以其實如果需要了解一下，推薦直接到官網去檢視

上一篇時也有提到說Kafka在0.8版本以前存在比較大的問題，1.x的算是目前生產環境中使用較多的版本

在quickStart就能看到相關的命令，比如

① 建立主題

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
將該命令修改一下

zookeeper localhost:2181 --replication-factor 2 --partitions 2 --topic tellYourDream
這時候就是zookeeper的地址為localhost:2181
兩個分割槽，兩個副本，一共4個副本，topic名稱為“tellYourDream”了
複製程式碼

還得注意，一般來說設定分割槽數建議是節點的倍數，這是為了讓服務節點分配均衡的舉措。

② 檢視主題：

bin/kafka-topics.sh --list --zookeeper localhost:2181
複製程式碼

③ 生產資訊

 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
This is a message
This is another message
複製程式碼

④ 消費資訊

 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
This is a message
This is another message
複製程式碼

這裡有個細節需要提及一下，就是我們0.8版本的kafka找的是zookeeper，zookeeper上確實是也存在著元資料資訊。

不過這存在著一些問題，zookeeper本身有一個過半服務的特性，這是一個限制，過半服務是指任何的請求都需要半數節點同意才能執行。每次有寫請求，它都要投票，因為它要保持資料的強一致性，做到節點狀態同步，所以高併發寫的效能不好。不適合做高併發的事。zookeeper是Kafka儲存元資料和交換叢集資訊的工具，主要是處理分散式一致性的問題。

⑤ 叢集測試

下面的命令就是生產50W條資料，每條資料200位元組，這條命令一執行就會產生一條報告，可以很直觀的看到叢集效能，看不懂的情況搜尋引擎也可以很好地幫助你解決問題

測試生產資料
bin/kafka-producer-perf-test.sh --topic test-topic --num-records 500000 --record-size 200 --throughput -1 --producer-props bootstrap.servers=hadoop03:9092,hadoop04:9092,hadoop05:9092 acks=-1
複製程式碼

每次消費2000條，叢集沒跑掛那就穩妥了

測試消費資料
bin/kafka-consumer-perf-test.sh --broker-list hadoop03:9092,hadoop53:9092 --fetch-size 2000 --messages 500000 --topic test-topic 
複製程式碼

四、KafkaManager

KafkaManager使用scala寫的專案。安裝步驟可以參考kafka叢集管理工具kafka-manager部署安裝，非常不錯。使用方法可以通過搜尋引擎查詢。

安裝好了之後可以使用jps命令檢視一下，會多出一個名字叫做ProdServerStart的服務

功能介紹

1. 管理多個kafka叢集
2. 便捷的檢查kafka叢集狀態(topics,brokers,備份分佈情況,分割槽分佈情況)
3. 選擇你要執行的副本
4. 基於當前分割槽狀況進行
5. 可以選擇topic配置並建立topic(0.8.1.1和0.8.2的配置不同)
6. 刪除topic(只支援0.8.2以上的版本並且要在broker配置中設定delete.topic.enable=true)
7. Topic list會指明哪些topic被刪除（在0.8.2以上版本適用）
8. 為已存在的topic增加分割槽
9. 為已存在的topic更新配置
10. 在多個topic上批量重分割槽
11. 在多個topic上批量重分割槽(可選partition broker位置)
複製程式碼

五、KafkaOffsetMonitor

KafkaOffsetMonitor就是一個jar包而已，是一個針對於消費者的工具。它可以用於監控消費延遲的問題，不過對於重複消費和訊息丟失等就無法解決，因為之後如果需要講解SparkStreaming，flink這些用於消費者的實踐的話，會使用到這個工具，所以現在先不展開，瞭解一下即可

啟動命令：

java -cp KafkaOffsetMonitor-assembly-0.3.0-SNAPSHOT.jar \
 com.quantifind.kafka.offsetapp.OffsetGetterWeb \
 --offsetStorage kafka \
 --zk xx:2181,xx:2181,xx:2181/kafka_cluster \
 --port 8088 \
 --refresh 60.seconds \
 --retain 2.days
複製程式碼

還有一些跨機房同步資料的像MirrorMaker這些，酌情使用。

finally

字數很多，還是希望可以耐心看完

之後會繼續展開生產者和消費者的原理，還有一些原始碼的講述。並且也會總結一些關於kafka經常遇到的一些經典問題提供出來（其實就是面試相關啦），感興趣的朋友可以持續關注。