Kafka基礎(chǔ)

消息系統(tǒng)的作用

應(yīng)該大部分小伙伴都清楚，用機(jī)油裝箱舉個(gè)例子。

所以消息系統(tǒng)就是如上圖我們所說(shuō)的倉(cāng)庫(kù)，能在中間過(guò)程作為緩存，并且實(shí)現(xiàn)解耦合的作用。

引入一個(gè)場(chǎng)景，我們知道中國(guó)移動(dòng)，中國(guó)聯(lián)通，中國(guó)電信的日志處理，是交給外包去做大數(shù)據(jù)分析的，假設(shè)現(xiàn)在它們的日志都交給了你做的系統(tǒng)去做用戶畫(huà)像分析。

按照剛剛前面提到的消息系統(tǒng)的作用，我們知道了消息系統(tǒng)其實(shí)就是一個(gè)模擬緩存，且僅僅是起到了緩存的作用而并不是真正的緩存，數(shù)據(jù)仍然是存儲(chǔ)在磁盤(pán)上面而不是內(nèi)存。

Topic主題

Kafka學(xué)習(xí)了數(shù)據(jù)庫(kù)里面的設(shè)計(jì)，在里面設(shè)計(jì)了topic(主題)，這個(gè)東西類(lèi)似于關(guān)系型數(shù)據(jù)庫(kù)的表。

此時(shí)我需要獲取中國(guó)移動(dòng)的數(shù)據(jù)，那就直接監(jiān)聽(tīng)TopicA即可。

Partition分區(qū)

kafka還有一個(gè)概念叫Partition(分區(qū))，分區(qū)具體在服務(wù)器上面表現(xiàn)起初就是一個(gè)目錄，一個(gè)主題下面有多個(gè)分區(qū)，這些分區(qū)會(huì)存儲(chǔ)到不同的服務(wù)器上面，或者說(shuō)，其實(shí)就是在不同的主機(jī)上建了不同的目錄。這些分區(qū)主要的信息就存在了.log文件里面。跟數(shù)據(jù)庫(kù)里面的分區(qū)差不多，是為了提高性能。

至于為什么提高了性能，很簡(jiǎn)單，多個(gè)分區(qū)多個(gè)線程，多個(gè)線程并行處理肯定會(huì)比單線程好得多。

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

注意：

分區(qū)會(huì)有單點(diǎn)故障問(wèn)題，所以我們會(huì)為每個(gè)分區(qū)設(shè)置副本數(shù);

分區(qū)的編號(hào)是從0開(kāi)始的。

Producer?-?生產(chǎn)者

往消息系統(tǒng)里面發(fā)送數(shù)據(jù)的就是生產(chǎn)者。

Consumer?-?消費(fèi)者

從Kafka里讀取數(shù)據(jù)的就是消費(fèi)者。

Message?-?消息

Kafka里面的我們處理的數(shù)據(jù)叫做消息。

Kafka的集群架構(gòu)

創(chuàng)建一個(gè)TopicA的主題，3個(gè)分區(qū)分別存儲(chǔ)在不同的服務(wù)器，也就是broker下面。Topic是一個(gè)邏輯上的概念，并不能直接在圖中把Topic的相關(guān)單元畫(huà)出。

需要注意：Kafka在0.8版本以前是沒(méi)有副本機(jī)制的，所以在面對(duì)服務(wù)器宕機(jī)的突發(fā)情況時(shí)會(huì)丟失數(shù)據(jù)，所以盡量避免使用這個(gè)版本之前的Kafka。

Replica?-?副本

Kafka中的partition為了保證數(shù)據(jù)安全，所以每個(gè)partition可以設(shè)置多個(gè)副本。

此時(shí)我們對(duì)分區(qū)0，1，2分別設(shè)置3個(gè)副本(其實(shí)設(shè)置兩個(gè)副本是比較合適的)。

而且其實(shí)每個(gè)副本都是有角色之分的，它們會(huì)選取一個(gè)副本作為leader，而其余的作為follower，我們的生產(chǎn)者在發(fā)送數(shù)據(jù)的時(shí)候，是直接發(fā)送到leader?partition里面，然后follower?partition會(huì)去leader那里自行同步數(shù)據(jù)，消費(fèi)者消費(fèi)數(shù)據(jù)的時(shí)候，也是從leader那去消費(fèi)數(shù)據(jù)的。

Consumer?Group?-?消費(fèi)者組

我們?cè)谙M(fèi)數(shù)據(jù)時(shí)會(huì)在代碼里面指定一個(gè)group.id，這個(gè)id代表的是消費(fèi)組的名字，而且這個(gè)group.id就算不設(shè)置，系統(tǒng)也會(huì)默認(rèn)設(shè)置。

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

我們所熟知的一些消息系統(tǒng)一般來(lái)說(shuō)會(huì)這樣設(shè)計(jì)，就是只要有一個(gè)消費(fèi)者去消費(fèi)了消息系統(tǒng)里面的數(shù)據(jù)，那么其余所有的消費(fèi)者都不能再去消費(fèi)這個(gè)數(shù)據(jù)?？墒荎afka并不是這樣，比如現(xiàn)在consumerA去消費(fèi)了一個(gè)topicA里面的數(shù)據(jù)。

consumerA:

group.id?=?a

consumerB:

group.id?=?a

consumerC:

group.id?=?b

consumerD:

group.id?=?b

再讓consumerB也去消費(fèi)TopicA的數(shù)據(jù)，它是消費(fèi)不到了，但是我們?cè)赾onsumerC中重新指定一個(gè)另外的group.id，consumerC是可以消費(fèi)到topicA的數(shù)據(jù)的。而consumerD也是消費(fèi)不到的，所以在Kafka中，不同組可有唯一的一個(gè)消費(fèi)者去消費(fèi)同一主題的數(shù)據(jù)。

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

consumer?group:a

consumerA

consumerB

consumerC

如圖，因?yàn)榍懊嫣岬竭^(guò)了消費(fèi)者會(huì)直接和leader建立聯(lián)系，所以它們分別消費(fèi)了三個(gè)leader，所以一個(gè)分區(qū)不會(huì)讓消費(fèi)者組里面的多個(gè)消費(fèi)者去消費(fèi)，但是在消費(fèi)者不飽和的情況下，一個(gè)消費(fèi)者是可以去消費(fèi)多個(gè)分區(qū)的數(shù)據(jù)的。

Controller

熟知一個(gè)規(guī)律：在大數(shù)據(jù)分布式文件系統(tǒng)里面，95%的都是主從式的架構(gòu)，個(gè)別是對(duì)等式的架構(gòu)，比如ElasticSearch。

Kafka也是主從式的架構(gòu)，主節(jié)點(diǎn)就叫controller，其余的為從節(jié)點(diǎn)，controller是需要和ZooKeeper進(jìn)行配合管理整個(gè)Kafka集群。

Kafka和ZooKeeper如何配合工作

Kafka嚴(yán)重依賴于ZooKeeper集群。所有的broker在啟動(dòng)的時(shí)候都會(huì)往ZooKeeper進(jìn)行注冊(cè)，目的就是選舉出一個(gè)controller，這個(gè)選舉過(guò)程非常簡(jiǎn)單粗暴，就是一個(gè)誰(shuí)先誰(shuí)當(dāng)?shù)倪^(guò)程，不涉及什么算法問(wèn)題。

那成為controller之后要做啥呢，它會(huì)監(jiān)聽(tīng)ZooKeeper里面的多個(gè)目錄，例如有一個(gè)目錄/brokers/，其他從節(jié)點(diǎn)往這個(gè)目錄上注冊(cè)(就是往這個(gè)目錄上創(chuàng)建屬于自己的子目錄而已)自己，這時(shí)命名規(guī)則一般是它們的id編號(hào)，比如/brokers/0，1，2。

注冊(cè)時(shí)各個(gè)節(jié)點(diǎn)必定會(huì)暴露自己的主機(jī)名，端口號(hào)等等的信息，此時(shí)controller就要去讀取注冊(cè)上來(lái)的從節(jié)點(diǎn)的數(shù)據(jù)(通過(guò)監(jiān)聽(tīng)機(jī)制)，生成集群的元數(shù)據(jù)信息，之后把這些信息都分發(fā)給其他的服務(wù)器，讓其他服務(wù)器能感知到集群中其它成員的存在。

此時(shí)模擬一個(gè)場(chǎng)景，我們創(chuàng)建一個(gè)主題(其實(shí)就是在ZooKeeper上/topics/topicA這樣創(chuàng)建一個(gè)目錄而已)，Kafka會(huì)把分區(qū)方案生成在這個(gè)目錄中，此時(shí)controller就監(jiān)聽(tīng)到了這一改變，它會(huì)去同步這個(gè)目錄的元信息，然后同樣下放給它的從節(jié)點(diǎn)，通過(guò)這個(gè)方法讓整個(gè)集群都得知這個(gè)分區(qū)方案，此時(shí)從節(jié)點(diǎn)就各自創(chuàng)建好目錄等待創(chuàng)建分區(qū)副本即可。這也是整個(gè)集群的管理機(jī)制。

加餐時(shí)間

Kafka性能好在什么地方？

順序?qū)?/p>

操作系統(tǒng)每次從磁盤(pán)讀寫(xiě)數(shù)據(jù)的時(shí)候，需要先尋址，也就是先要找到數(shù)據(jù)在磁盤(pán)上的物理位置，然后再進(jìn)行數(shù)據(jù)讀寫(xiě)，如果是機(jī)械硬盤(pán)，尋址就需要較長(zhǎng)的時(shí)間。

Kafka的設(shè)計(jì)中，數(shù)據(jù)其實(shí)是存儲(chǔ)在磁盤(pán)上面，一般來(lái)說(shuō)，會(huì)把數(shù)據(jù)存儲(chǔ)在內(nèi)存上面性能才會(huì)好。但是Kafka用的是順序?qū)懀芳訑?shù)據(jù)是追加到末尾，磁盤(pán)順序?qū)懙男阅軜O高，在磁盤(pán)個(gè)數(shù)一定，轉(zhuǎn)數(shù)達(dá)到一定的情況下，基本和內(nèi)存速度一致。

隨機(jī)寫(xiě)的話是在文件的某個(gè)位置修改數(shù)據(jù)，性能會(huì)較低。

零拷貝

先來(lái)看看非零拷貝的情況。

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

日志分段存儲(chǔ)

Kafka規(guī)定了一個(gè)分區(qū)內(nèi)的.log文件最大為1G，做這個(gè)限制目的是為了方便把.log加載到內(nèi)存去操作。

這個(gè)9936472之類(lèi)的數(shù)字，就是代表了這個(gè)日志段文件里包含的起始o(jì)ffset，也就說(shuō)明這個(gè)分區(qū)里至少都寫(xiě)入了接近1000萬(wàn)條數(shù)據(jù)了。Kafka?broker有一個(gè)參數(shù)，log.segment.bytes，限定了每個(gè)日志段文件的大小，最大就是1GB，一個(gè)日志段文件滿了，就自動(dòng)開(kāi)一個(gè)新的日志段文件來(lái)寫(xiě)入，避免單個(gè)文件過(guò)大，影響文件的讀寫(xiě)性能，這個(gè)過(guò)程叫做log?rolling，正在被寫(xiě)入的那個(gè)日志段文件，叫做active?log?segment。

如果大家有看前面的兩篇有關(guān)于HDFS的文章時(shí)，就會(huì)發(fā)現(xiàn)NameNode的edits?log也會(huì)做出限制，所以這些框架都是會(huì)考慮到這些問(wèn)題。

Kafka的網(wǎng)絡(luò)設(shè)計(jì)

Kafka的網(wǎng)絡(luò)設(shè)計(jì)和Kafka的調(diào)優(yōu)有關(guān)，這也是為什么它能支持高并發(fā)的原因。

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

線程池里面默認(rèn)有8個(gè)線程，這些線程是用來(lái)處理request的，解析請(qǐng)求，如果request是寫(xiě)請(qǐng)求，就寫(xiě)到磁盤(pán)里。讀的話返回結(jié)果。

processor會(huì)從response中讀取響應(yīng)數(shù)據(jù)，然后再返回給客戶端。這就是Kafka的網(wǎng)絡(luò)三層架構(gòu)。

所以如果我們需要對(duì)Kafka進(jìn)行增強(qiáng)調(diào)優(yōu)，增加processor并增加線程池里面的處理線程，就可以達(dá)到效果。request和response那一塊部分其實(shí)就是起到了一個(gè)緩存的效果，是考慮到processor們生成請(qǐng)求太快，線程數(shù)不夠不能及時(shí)處理的問(wèn)題。

所以這就是一個(gè)加強(qiáng)版的reactor網(wǎng)絡(luò)線程模型。

Kafka的生產(chǎn)集群部署

方案背景

假設(shè)每天集群需要承載10億數(shù)據(jù)。一天24小時(shí)，晚上12點(diǎn)到凌晨8點(diǎn)幾乎沒(méi)多少數(shù)據(jù)。

使用二八法則估計(jì)，也就是80%的數(shù)據(jù)(8億)會(huì)在16個(gè)小時(shí)涌入，而且8億的80%的數(shù)據(jù)(6.4億)會(huì)在這16個(gè)小時(shí)的20%時(shí)間(3小時(shí))涌入。

QPS計(jì)算公式：640000000?÷?(3x60x60)?=?60000，也就是說(shuō)高峰期的時(shí)候Kafka集群要扛住每秒6萬(wàn)的并發(fā)。

磁盤(pán)空間計(jì)算，每天10億數(shù)據(jù)，每條50kb，也就是46T的數(shù)據(jù)。保存2個(gè)副本(在上一篇中也提到過(guò)其實(shí)兩個(gè)副本會(huì)比較好，因?yàn)閒ollower需要去leader那里同步數(shù)據(jù)，同步數(shù)據(jù)的過(guò)程需要耗費(fèi)網(wǎng)絡(luò)，而且需要磁盤(pán)空間，但是這個(gè)需要根據(jù)實(shí)際情況考慮)，46?*?2?=?92T，保留最近3天的數(shù)據(jù)。故需要?92?*?3?=?276T。

QPS方面

部署Kafka，Hadoop，MySQL……等核心分布式系統(tǒng)，一般建議直接采用物理機(jī)，拋棄使用一些低配置的虛擬機(jī)的想法。高并發(fā)這個(gè)東西，不可能是說(shuō)，你需要支撐6萬(wàn)QPS，你的集群就剛好把這6萬(wàn)并發(fā)卡的死死的。假如某一天出一些活動(dòng)讓數(shù)據(jù)量瘋狂上漲，那整個(gè)集群就會(huì)垮掉。

但是，假如說(shuō)你只要支撐6w?QPS，單臺(tái)物理機(jī)本身就能扛住4~5萬(wàn)的并發(fā)。所以這時(shí)2臺(tái)物理機(jī)絕對(duì)絕對(duì)夠了。但是這里有一個(gè)問(wèn)題，我們通常是建議，公司預(yù)算充足，盡量是讓高峰QPS控制在集群能承載的總QPS的30%左右(也就是集群的處理能力是高峰期的3~4倍這個(gè)樣子)，所以我們搭建的kafka集群能承載的總QPS為20萬(wàn)~30萬(wàn)才是安全的。所以大體上來(lái)說(shuō)，需要5~7臺(tái)物理機(jī)來(lái)部署，基本上就很安全了，每臺(tái)物理機(jī)要求吞吐量在每秒4~5萬(wàn)條數(shù)據(jù)就可以了，物理機(jī)的配置和性能也不需要特別高。

磁盤(pán)方面

磁盤(pán)數(shù)量

需要5臺(tái)物理機(jī)的情況，需要存儲(chǔ)276T的數(shù)據(jù)，平均下來(lái)差不多一臺(tái)56T的數(shù)據(jù)。這個(gè)具體看磁盤(pán)數(shù)和盤(pán)的大小。

SAS還是SSD

現(xiàn)在我們需要考慮一個(gè)問(wèn)題：是需要SSD固態(tài)硬盤(pán)，還是普通機(jī)械硬盤(pán)？

SSD就是固態(tài)硬盤(pán)，比機(jī)械硬盤(pán)要快，那么到底是快在哪里呢？其實(shí)SSD的快主要是快在磁盤(pán)隨機(jī)讀寫(xiě)，就要對(duì)磁盤(pán)上的隨機(jī)位置來(lái)讀寫(xiě)的時(shí)候，SSD比機(jī)械硬盤(pán)要快。比如說(shuō)MySQL這種就應(yīng)該使用SSD了(MySQL需要隨機(jī)讀寫(xiě))。比如說(shuō)我們?cè)谝?guī)劃和部署線上系統(tǒng)的MySQL集群的時(shí)候，一般來(lái)說(shuō)必須用SSD，性能可以提高很多，這樣MySQL可以承載的并發(fā)請(qǐng)求量也會(huì)高很多，而且SQL語(yǔ)句執(zhí)行的性能也會(huì)提高很多。

因?yàn)閷?xiě)磁盤(pán)的時(shí)候Kafka是順序?qū)懙?。機(jī)械硬盤(pán)順序?qū)懙男阅軝C(jī)會(huì)跟內(nèi)存讀寫(xiě)的性能是差不多的，所以對(duì)于Kafka集群來(lái)說(shuō)其實(shí)使用機(jī)械硬盤(pán)就可以了。如果是需要自己創(chuàng)業(yè)或者是在公司成本不足的情況下，經(jīng)費(fèi)是能夠縮減就盡量縮減的。

內(nèi)存角度

JVM非常怕出現(xiàn)full?gc的情況。Kafka自身的JVM是用不了過(guò)多堆內(nèi)存的，因?yàn)镵afka設(shè)計(jì)就是規(guī)避掉用JVM對(duì)象來(lái)保存數(shù)據(jù)，避免頻繁full?gc導(dǎo)致的問(wèn)題，所以一般Kafka自身的JVM堆內(nèi)存，分配個(gè)10G左右就夠了，剩下的內(nèi)存全部留給OS?cache。

那服務(wù)器需要多少內(nèi)存呢。我們估算一下，大概有100個(gè)topic，所以要保證有100個(gè)topic的leader?partition的數(shù)據(jù)在操作系統(tǒng)的內(nèi)存里。100個(gè)topic，一個(gè)topic有5個(gè)partition。那么總共會(huì)有500個(gè)partition。每個(gè)partition的大小是1G(在上一篇中的日志分段存儲(chǔ)中規(guī)定了.log文件不能超過(guò)1個(gè)G)，我們有2個(gè)副本，也就是說(shuō)要把100個(gè)topic的leader?partition數(shù)據(jù)都駐留在內(nèi)存里需要1000G的內(nèi)存。

我們現(xiàn)在有5臺(tái)服務(wù)器，所以平均下來(lái)每天服務(wù)器需要200G的內(nèi)存，但是其實(shí)partition的數(shù)據(jù)我們沒(méi)必要所有的都要駐留在內(nèi)存里面，只需要25%的數(shù)據(jù)在內(nèi)存就行，200G?*?0.25?=?50G就可以了(因?yàn)樵诩褐械纳a(chǎn)者和消費(fèi)者幾乎也算是實(shí)時(shí)的，基本不會(huì)出現(xiàn)消息積壓太多的情況)。所以一共需要60G(附帶上剛剛的10G?Kafka服務(wù))的內(nèi)存，故我們可以挑選64G內(nèi)存的服務(wù)器也行，大不了partition的數(shù)據(jù)再少一點(diǎn)在內(nèi)存，當(dāng)然如果能夠提供128G內(nèi)存那就更好。

CPU?core

CPU規(guī)劃，主要是看你的這個(gè)進(jìn)程里會(huì)有多少個(gè)線程，線程主要是依托多核CPU來(lái)執(zhí)行的，如果你的線程特別多，但是CPU核很少，就會(huì)導(dǎo)致你的CPU負(fù)載很高，會(huì)導(dǎo)致整體工作線程執(zhí)行的效率不太高，上一篇的Kafka的網(wǎng)絡(luò)設(shè)計(jì)中講過(guò)Kafka的Broker的模型。acceptor線程負(fù)責(zé)去接入客戶端的連接請(qǐng)求，但是他接入了之后其實(shí)就會(huì)把連接分配給多個(gè)processor，默認(rèn)是3個(gè)，但是一般生產(chǎn)環(huán)境建議大家還是多加幾個(gè)，整體可以提升kafka的吞吐量比如說(shuō)你可以增加到6個(gè)，或者是9個(gè)。另外就是負(fù)責(zé)處理請(qǐng)求的線程，是一個(gè)線程池，默認(rèn)是8個(gè)線程，在生產(chǎn)集群里，建議大家可以把這塊的線程數(shù)量稍微多加個(gè)2倍~3倍，其實(shí)都正常，比如說(shuō)搞個(gè)16個(gè)工作線程，24個(gè)工作線程。

后臺(tái)會(huì)有很多的其他的一些線程，比如說(shuō)定期清理7天前數(shù)據(jù)的線程，Controller負(fù)責(zé)感知和管控整個(gè)集群的線程，副本同步拉取數(shù)據(jù)的線程，這樣算下來(lái)每個(gè)broker起碼會(huì)有上百個(gè)線程。根據(jù)經(jīng)驗(yàn)4個(gè)CPU?core，一般來(lái)說(shuō)幾十個(gè)線程，在高峰期CPU幾乎都快打滿了。8個(gè)CPU?core，也就能夠比較寬裕的支撐幾十個(gè)線程繁忙的工作。所以Kafka的服務(wù)器一般是建議16核，基本上可以hold住一兩百線程的工作。當(dāng)然如果可以給到32?CPU?core那就最好不過(guò)了。

網(wǎng)卡

現(xiàn)在的網(wǎng)基本就是千兆網(wǎng)卡(1GB?/?s)，還有萬(wàn)兆網(wǎng)卡(10GB?/?s)。kafka集群之間，broker和broker之間是會(huì)做數(shù)據(jù)同步的，因?yàn)閘eader要同步數(shù)據(jù)到follower上去，他們是在不同的broker機(jī)器上的，broker機(jī)器之間會(huì)進(jìn)行頻繁的數(shù)據(jù)同步，傳輸大量的數(shù)據(jù)。那每秒兩臺(tái)broker機(jī)器之間大概會(huì)傳輸多大的數(shù)據(jù)量？

高峰期每秒大概會(huì)涌入6萬(wàn)條數(shù)據(jù)，約每天處理10000個(gè)請(qǐng)求，每個(gè)請(qǐng)求50kb，故每秒約進(jìn)來(lái)488M數(shù)據(jù)，我們還有副本同步數(shù)據(jù)，故高峰期的時(shí)候需要488M?*?2?=?976M/s的網(wǎng)絡(luò)帶寬，所以在高峰期的時(shí)候，使用千兆帶寬，網(wǎng)絡(luò)還是非常有壓力的。

綜上描述

10億數(shù)據(jù)，6w/s的吞吐量，276T的數(shù)據(jù)，5臺(tái)物理機(jī)

硬盤(pán)：11(SAS)?*?7T，7200轉(zhuǎn)

內(nèi)存：64GB/128GB，JVM分配10G，剩余的給os?cache

CPU：16核/32核

網(wǎng)絡(luò)：千兆網(wǎng)卡，萬(wàn)兆更好

Kafka的集群搭建

進(jìn)到Kafka的config文件夾下，會(huì)發(fā)現(xiàn)有很多很多的配置文件，可是都不需要你來(lái)修改，你僅僅需要點(diǎn)開(kāi)一個(gè)叫作server.properties的文件就夠了。

【broker.id】

每個(gè)broker都必須自己設(shè)置的一個(gè)唯一id，可以在0~255之間

【log.dirs】

這個(gè)極為重要，Kafka的所有數(shù)據(jù)就是寫(xiě)入這個(gè)目錄下的磁盤(pán)文件中的，如果說(shuō)機(jī)器上有多塊物理硬盤(pán)，那么可以把多個(gè)目錄掛載到不同的物理硬盤(pán)上，然后這里可以設(shè)置多個(gè)目錄，這樣Kafka可以數(shù)據(jù)分散到多塊物理硬盤(pán)，多個(gè)硬盤(pán)的磁頭可以并行寫(xiě)，這樣可以提升吞吐量。ps：多個(gè)目錄用英文逗號(hào)分隔

【zookeeper.connect】

連接Kafka底層的ZooKeeper集群的

【Listeners】

broker監(jiān)聽(tīng)客戶端發(fā)起請(qǐng)求的端口號(hào)，默認(rèn)是9092

【num.network.threads】默認(rèn)值為3

【num.io.threads】默認(rèn)值為8

細(xì)心的朋友們應(yīng)該已經(jīng)發(fā)現(xiàn)了，這就是上一篇我們?cè)诰W(wǎng)絡(luò)架構(gòu)上提到的processor和處理線程池的線程數(shù)目。

所以說(shuō)掌握Kafka網(wǎng)絡(luò)架構(gòu)顯得尤為重要。

現(xiàn)在你看到這兩個(gè)參數(shù)，就知道這就是Kafka集群性能的關(guān)鍵參數(shù)了

【unclean.leader.election.enable】

默認(rèn)是?false，意思就是只能選舉ISR列表里的follower成為新的leader，1.0版本后才設(shè)為?false，之前都是?true，允許非ISR列表的follower選舉為新的leader

【delete.topic.enable】

默認(rèn)?true，允許刪除topic

【log.retention.hours】

可以設(shè)置一下，要保留數(shù)據(jù)多少個(gè)小時(shí)，這個(gè)就是底層的磁盤(pán)文件，默認(rèn)保留7天的數(shù)據(jù)，根據(jù)自己的需求來(lái)就行了

【min.insync.replicas】

acks=-1(一條數(shù)據(jù)必須寫(xiě)入ISR里所有副本才算成功)，你寫(xiě)一條數(shù)據(jù)只要寫(xiě)入leader就算成功了，不需要等待同步到follower才算寫(xiě)成功。但是此時(shí)如果一個(gè)follower宕機(jī)了，你寫(xiě)一條數(shù)據(jù)到leader之后，leader也宕機(jī)，會(huì)導(dǎo)致數(shù)據(jù)的丟失。

因?yàn)閷?shí)際的集群搭建說(shuō)真的沒(méi)有太大難度，所以搭建的過(guò)程就不詳細(xì)展開(kāi)了，網(wǎng)上應(yīng)該很多相關(guān)資料。

Kafka的簡(jiǎn)單集群操作

在操作Kafka集群的時(shí)候，不同的Kafka版本命令的寫(xiě)法是不一樣的，所以其實(shí)如果需要了解一下，推薦直接到官網(wǎng)去查看。

上一篇?時(shí)也有提到說(shuō)Kafka在0.8版本以前存在比較大的問(wèn)題，1.x的算是目前生產(chǎn)環(huán)境中使用較多的版本。

在quickStart就能看到相關(guān)的命令，比如：

創(chuàng)建主題

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

這時(shí)候就是zookeeper的地址為localhost:2181

兩個(gè)分區(qū)，兩個(gè)副本，一共4個(gè)副本，topic名稱為“tellYourDream”了

還得注意，一般來(lái)說(shuō)設(shè)置分區(qū)數(shù)建議是節(jié)點(diǎn)的倍數(shù)，這是為了讓服務(wù)節(jié)點(diǎn)分配均衡的舉措。

查看主題

bin/kafka-topics.sh?--list?--zookeeper?localhost:2181

生產(chǎn)信息

bin/kafka-console-producer.sh?--broker-list?localhost:9092?--topic?test

This?is?a?message

This?is?another?message

消費(fèi)信息

bin/kafka-console-consumer.sh?--bootstrap-server?localhost:9092?--topic?test--from-beginning

This?is?a?message

This?is?another?message

這里有個(gè)細(xì)節(jié)需要提及一下，就是我們0.8版本的Kafka找的是ZooKeeper，ZooKeeper上確實(shí)是也存在著元數(shù)據(jù)信息。

不過(guò)這存在著一些問(wèn)題，ZooKeeper本身有一個(gè)過(guò)半服務(wù)的特性，這是一個(gè)限制，過(guò)半服務(wù)是指任何的請(qǐng)求都需要半數(shù)節(jié)點(diǎn)同意才能執(zhí)行。每次有寫(xiě)請(qǐng)求，它都要投票，因?yàn)樗３謹(jǐn)?shù)據(jù)的強(qiáng)一致性，做到節(jié)點(diǎn)狀態(tài)同步，所以高并發(fā)寫(xiě)的性能不好。不適合做高并發(fā)的事。ZooKeeper是Kafka存儲(chǔ)元數(shù)據(jù)和交換集群信息的工具，主要是處理分布式一致性的問(wèn)題。

集群測(cè)試

下面的命令就是生產(chǎn)50W條數(shù)據(jù)，每條數(shù)據(jù)200字節(jié)，這條命令一運(yùn)行就會(huì)產(chǎn)生一條報(bào)告，可以很直觀的看到集群性能，看不懂的情況搜索引擎也可以很好地幫助你解決問(wèn)題。

每次消費(fèi)2000條，集群沒(méi)跑掛那就穩(wěn)妥了。

KafkaManager

KafkaManager使用scala寫(xiě)的項(xiàng)目，非常不錯(cuò)。使用方法可以通過(guò)搜索引擎查找。

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

安裝好了之后可以使用jps命令查看一下，會(huì)多出一個(gè)名字叫做ProdServerStart的服務(wù)。

功能介紹：

管理多個(gè)Kafka集群

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

選擇你要運(yùn)行的副本

基于當(dāng)前分區(qū)狀況進(jìn)行

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

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

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

為已存在的topic增加分區(qū)

為已存在的topic更新配置

在多個(gè)topic上批量重分區(qū)

在多個(gè)topic上批量重分區(qū)(可選partition?broker位置)

KafkaOffsetMonitor

KafkaOffsetMonitor就是一個(gè)jar包而已，是一個(gè)針對(duì)于消費(fèi)者的工具。它可以用于監(jiān)控消費(fèi)延遲的問(wèn)題，不過(guò)對(duì)于重復(fù)消費(fèi)和消息丟失等就無(wú)法解決，因?yàn)橹笕绻枰v解SparkStreaming，flink這些用于消費(fèi)者的實(shí)踐的話，會(huì)使用到這個(gè)工具，所以現(xiàn)在先不展開(kāi)，了解一下即可。

啟動(dòng)命令：

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

還有一些跨機(jī)房同步數(shù)據(jù)的像MirrorMaker這些，酌情使用。

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