Kafka基礎

消息系統的作用

應該大部分小伙伴都清楚，用機油裝箱舉個例子。

所以消息系統就是如上圖我們所說的倉庫，能在中間過程作為緩存，并且實現解耦合的作用。

引入一個場景，我們知道中國移動，中國聯通，中國電信的日志處理，是交給外包去做大數據分析的，假設現在它們的日志都交給了你做的系統去做用戶畫像分析。

按照剛剛前面提到的消息系統的作用，我們知道了消息系統其實就是一個模擬緩存，且僅僅是起到了緩存的作用而并不是真正的緩存，數據仍然是存儲在磁盤上面而不是內存。

Topic主題

Kafka學習了數據庫里面的設計，在里面設計了topic(主題)，這個東西類似于關系型數據庫的表。

此時我需要獲取中國移動的數據，那就直接監聽TopicA即可。

Partition分區

kafka還有一個概念叫Partition(分區)，分區具體在服務器上面表現起初就是一個目錄，一個主題下面有多個分區，這些分區會存儲到不同的服務器上面，或者說，其實就是在不同的主機上建了不同的目錄。這些分區主要的信息就存在了.log文件里面。跟數據庫里面的分區差不多，是為了提高性能。

至于為什么提高了性能，很簡單，多個分區多個線程，多個線程并行處理肯定會比單線程好得多。

Topic和partition像是HBASE里的table和region的概念，table只是一個邏輯上的概念，真正存儲數據的是region，這些region會分布式地存儲在各個服務器上面，對應于Kafka，也是一樣，Topic也是邏輯概念，而partition就是分布式存儲單元。這個設計是保證了海量數據處理的基礎。我們可以對比一下，如果HDFS沒有block的設計，一個100T的文件也只能單獨放在一個服務器上面，那就直接占滿整個服務器了，引入block后，大文件可以分散存儲在不同的服務器上。

注意：

分區會有單點故障問題，所以我們會為每個分區設置副本數;

分區的編號是從0開始的。

Producer?-?生產者

往消息系統里面發送數據的就是生產者。

Consumer?-?消費者

從Kafka里讀取數據的就是消費者。

Message?-?消息

Kafka里面的我們處理的數據叫做消息。

Kafka的集群架構

創建一個TopicA的主題，3個分區分別存儲在不同的服務器，也就是broker下面。Topic是一個邏輯上的概念，并不能直接在圖中把Topic的相關單元畫出。

需要注意：Kafka在0.8版本以前是沒有副本機制的，所以在面對服務器宕機的突發情況時會丟失數據，所以盡量避免使用這個版本之前的Kafka。

Replica?-?副本

Kafka中的partition為了保證數據安全，所以每個partition可以設置多個副本。

此時我們對分區0，1，2分別設置3個副本(其實設置兩個副本是比較合適的)。

而且其實每個副本都是有角色之分的，它們會選取一個副本作為leader，而其余的作為follower，我們的生產者在發送數據的時候，是直接發送到leader?partition里面，然后follower?partition會去leader那里自行同步數據，消費者消費數據的時候，也是從leader那去消費數據的。

Consumer?Group?-?消費者組

我們在消費數據時會在代碼里面指定一個group.id，這個id代表的是消費組的名字，而且這個group.id就算不設置，系統也會默認設置。

conf.setProperty(?"group.id",?"tellYourDream")

我們所熟知的一些消息系統一般來說會這樣設計，就是只要有一個消費者去消費了消息系統里面的數據，那么其余所有的消費者都不能再去消費這個數據。可是Kafka并不是這樣，比如現在consumerA去消費了一個topicA里面的數據。

consumerA:

group.id?=?a

consumerB:

group.id?=?a

consumerC:

group.id?=?b

consumerD:

group.id?=?b

再讓consumerB也去消費TopicA的數據，它是消費不到了，但是我們在consumerC中重新指定一個另外的group.id，consumerC是可以消費到topicA的數據的。而consumerD也是消費不到的，所以在Kafka中，不同組可有唯一的一個消費者去消費同一主題的數據。

所以消費者組就是讓多個消費者并行消費信息而存在的，而且它們不會消費到同一個消息，如下，consumerA，B，C是不會互相干擾的。

consumer?group:a

consumerA

consumerB

consumerC

如圖，因為前面提到過了消費者會直接和leader建立聯系，所以它們分別消費了三個leader，所以一個分區不會讓消費者組里面的多個消費者去消費，但是在消費者不飽和的情況下，一個消費者是可以去消費多個分區的數據的。

Controller

熟知一個規律：在大數據分布式文件系統里面，95%的都是主從式的架構，個別是對等式的架構，比如ElasticSearch。

Kafka也是主從式的架構，主節點就叫controller，其余的為從節點，controller是需要和ZooKeeper進行配合管理整個Kafka集群。

Kafka和ZooKeeper如何配合工作

Kafka嚴重依賴于ZooKeeper集群。所有的broker在啟動的時候都會往ZooKeeper進行注冊，目的就是選舉出一個controller，這個選舉過程非常簡單粗暴，就是一個誰先誰當的過程，不涉及什么算法問題。

那成為controller之后要做啥呢，它會監聽ZooKeeper里面的多個目錄，例如有一個目錄/brokers/，其他從節點往這個目錄上注冊(就是往這個目錄上創建屬于自己的子目錄而已)自己，這時命名規則一般是它們的id編號，比如/brokers/0，1，2。

注冊時各個節點必定會暴露自己的主機名，端口號等等的信息，此時controller就要去讀取注冊上來的從節點的數據(通過監聽機制)，生成集群的元數據信息，之后把這些信息都分發給其他的服務器，讓其他服務器能感知到集群中其它成員的存在。

此時模擬一個場景，我們創建一個主題(其實就是在ZooKeeper上/topics/topicA這樣創建一個目錄而已)，Kafka會把分區方案生成在這個目錄中，此時controller就監聽到了這一改變，它會去同步這個目錄的元信息，然后同樣下放給它的從節點，通過這個方法讓整個集群都得知這個分區方案，此時從節點就各自創建好目錄等待創建分區副本即可。這也是整個集群的管理機制。

加餐時間

Kafka性能好在什么地方？

順序寫

操作系統每次從磁盤讀寫數據的時候，需要先尋址，也就是先要找到數據在磁盤上的物理位置，然后再進行數據讀寫，如果是機械硬盤，尋址就需要較長的時間。

Kafka的設計中，數據其實是存儲在磁盤上面，一般來說，會把數據存儲在內存上面性能才會好。但是Kafka用的是順序寫，追加數據是追加到末尾，磁盤順序寫的性能極高，在磁盤個數一定，轉數達到一定的情況下，基本和內存速度一致。

隨機寫的話是在文件的某個位置修改數據，性能會較低。

零拷貝

先來看看非零拷貝的情況。

可以看到數據的拷貝從內存拷貝到Kafka服務進程那塊，又拷貝到socket緩存那塊，整個過程耗費的時間比較高，Kafka利用了Linux的sendFile技術(NIO)，省去了進程切換和一次數據拷貝，讓性能變得更好。

日志分段存儲

Kafka規定了一個分區內的.log文件最大為1G，做這個限制目的是為了方便把.log加載到內存去操作。

這個9936472之類的數字，就是代表了這個日志段文件里包含的起始offset，也就說明這個分區里至少都寫入了接近1000萬條數據了。Kafka?broker有一個參數，log.segment.bytes，限定了每個日志段文件的大小，最大就是1GB，一個日志段文件滿了，就自動開一個新的日志段文件來寫入，避免單個文件過大，影響文件的讀寫性能，這個過程叫做log?rolling，正在被寫入的那個日志段文件，叫做active?log?segment。

如果大家有看前面的兩篇有關于HDFS的文章時，就會發現NameNode的edits?log也會做出限制，所以這些框架都是會考慮到這些問題。

Kafka的網絡設計

Kafka的網絡設計和Kafka的調優有關，這也是為什么它能支持高并發的原因。

首先客戶端發送請求全部會先發送給一個Acceptor，broker里面會存在3個線程(默認是3個)，這3個線程都是叫做processor，Acceptor不會對客戶端的請求做任何的處理，直接封裝成一個個socketChannel發送給這些processor形成一個隊列，發送的方式是輪詢，就是先給第一個processor發送，然后再給第二個，第三個，然后又回到第一個。消費者線程去消費這些socketChannel時，會獲取一個個request請求，這些request請求中就會伴隨著數據。

線程池里面默認有8個線程，這些線程是用來處理request的，解析請求，如果request是寫請求，就寫到磁盤里。讀的話返回結果。

processor會從response中讀取響應數據，然后再返回給客戶端。這就是Kafka的網絡三層架構。

所以如果我們需要對Kafka進行增強調優，增加processor并增加線程池里面的處理線程，就可以達到效果。request和response那一塊部分其實就是起到了一個緩存的效果，是考慮到processor們生成請求太快，線程數不夠不能及時處理的問題。

所以這就是一個加強版的reactor網絡線程模型。

Kafka的生產集群部署

方案背景

假設每天集群需要承載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。

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萬條數據就可以了，物理機的配置和性能也不需要特別高。

磁盤方面

磁盤數量

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

SAS還是SSD

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

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

因為寫磁盤的時候Kafka是順序寫的。機械硬盤順序寫的性能機會跟內存讀寫的性能是差不多的，所以對于Kafka集群來說其實使用機械硬盤就可以了。如果是需要自己創業或者是在公司成本不足的情況下，經費是能夠縮減就盡量縮減的。

內存角度

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內存那就更好。

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那就最好不過了。

網卡

現在的網基本就是千兆網卡(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字節，這條命令一運行就會產生一條報告，可以很直觀的看到集群性能，看不懂的情況搜索引擎也可以很好地幫助你解決問題。

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

KafkaManager

KafkaManager使用scala寫的項目，非常不錯。使用方法可以通過搜索引擎查找。

安裝步驟可以參考：https://www.cnblogs.com/dadonggg/p/8205302.html

安裝好了之后可以使用jps命令查看一下，會多出一個名字叫做ProdServerStart的服務。

功能介紹：

管理多個Kafka集群

便捷的檢查Kafka集群狀態(topics，brokers，備份分布情況，分區分布情況)

選擇你要運行的副本

基于當前分區狀況進行

可以選擇topic配置并創建topic(0.8.1.1和0.8.2的配置不同)

刪除topic(只支持0.8.2以上的版本并且要在broker配置中設置delete.topic.enable=true)

Topic?list會指明哪些topic被刪除(在0.8.2以上版本適用)

為已存在的topic增加分區

為已存在的topic更新配置

在多個topic上批量重分區

在多個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這些，酌情使用。

最后，小編給您推薦，金山毒霸“文檔保護”，文檔自動備份，防止重要文件丟失，攔截惡意篡改，是您文檔的好幫手。