一、Kafka簡(jiǎn)介

Kafka是什么

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)（消息引擎系統(tǒng)），它可以處理消費(fèi)者在網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)。 這種動(dòng)作（網(wǎng)頁瀏覽，
搜索和其他用戶的行動(dòng)）是在現(xiàn)代網(wǎng)絡(luò)上的許多社會(huì)功能的一個(gè)關(guān)鍵因素。 這些數(shù)據(jù)通常是由于吞吐量的要求而通過處理日志和日志聚合來
解決。 對(duì)于像Hadoop一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。Kafka的目的是通過Hadoop
的并行加載機(jī)制來統(tǒng)一線上和離線的消息處理，也是為了通過集群來提供實(shí)時(shí)的消息。

其實(shí)我們簡(jiǎn)單點(diǎn)理解就是系統(tǒng)A發(fā)送消息給kafka（消息引擎系統(tǒng)），系統(tǒng)B從kafka中讀取A發(fā)送的消息。而kafka就是個(gè)中間商。

1.1 Kafka的特性:

• 高吞吐量、低延遲：kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒，每個(gè)topic可以分多個(gè)partition, consumer group 對(duì)partition進(jìn)行consume操作。

• 可擴(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í)讀寫

1.2 Kafka的使用場(chǎng)景：

Kafaka經(jīng)常用于削峰、解耦、異步。

• 日志收集：一個(gè)公司可以用Kafka可以收集各種服務(wù)的log，通過kafka以統(tǒng)一接口服務(wù)的方式開放給各種consumer，例如hadoop、Hbase、Solr等。

• 消息系統(tǒng)：解耦生產(chǎn)者和消費(fèi)者、緩存消息等。

• 用戶活動(dòng)跟蹤：Kafka經(jīng)常被用來記錄web用戶或者app用戶的各種活動(dòng)，如瀏覽網(wǎng)頁、搜索、點(diǎn)擊等活動(dòng)，這些活動(dòng)信息被各個(gè)服務(wù)器發(fā)布到kafka的topic中，然后訂閱者通過訂閱這些topic來做實(shí)時(shí)的監(jiān)控分析，或者裝載到hadoop、數(shù)據(jù)倉(cāng)庫中做離線分析和挖掘。

• 運(yùn)營(yíng)指標(biāo)：Kafka也經(jīng)常用來記錄運(yùn)營(yíng)監(jiān)控?cái)?shù)據(jù)。包括收集各種分布式應(yīng)用的數(shù)據(jù)，生產(chǎn)各種操作的集中反饋，比如報(bào)警和報(bào)告。

• 流式處理：比如spark streaming和storm

• 事件源

1.3 Kakfa的設(shè)計(jì)思想

Kakfa Broker Leader的選舉：Kakfa Broker集群受Zookeeper管理。所有的Kafka Broker節(jié)點(diǎn)一起去Zookeeper上注冊(cè)一個(gè)臨時(shí)節(jié)點(diǎn)，因?yàn)橹挥幸粋€(gè)Kafka Broker會(huì)注冊(cè)成功，其他的都會(huì)失敗，所以這個(gè)成功在Zookeeper上注冊(cè)臨時(shí)節(jié)點(diǎn)的這個(gè)Kafka Broker會(huì)成為Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。（這個(gè)過程叫Controller在ZooKeeper注冊(cè)Watch）。這個(gè)Controller會(huì)監(jiān)聽其他的Kafka Broker的所有信息，如果這個(gè)kafka broker controller宕機(jī)了，在zookeeper上面的那個(gè)臨時(shí)節(jié)點(diǎn)就會(huì)消失，此時(shí)所有的kafka broker又會(huì)一起去Zookeeper上注冊(cè)一個(gè)臨時(shí)節(jié)點(diǎn)，因?yàn)橹挥幸粋€(gè)Kafka Broker會(huì)注冊(cè)成功，其他的都會(huì)失敗，所以這個(gè)成功在Zookeeper上注冊(cè)臨時(shí)節(jié)點(diǎn)的這個(gè)Kafka Broker會(huì)成為Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。例如：一旦有一個(gè)broker宕機(jī)了，這個(gè)kafka broker controller會(huì)讀取該宕機(jī)broker上所有的partition在zookeeper上的狀態(tài)，并選取ISR列表中的一個(gè)replica作為partition leader（如果ISR列表中的replica全掛，選一個(gè)幸存的replica作為leader; 如果該partition的所有的replica都宕機(jī)了，則將新的leader設(shè)置為-1，等待恢復(fù)，等待ISR中的任一個(gè)Replica“活”過來，并且選它作為L(zhǎng)eader；或選擇第一個(gè)“活”過來的Replica（不一定是ISR中的）作為L(zhǎng)eader），這個(gè)broker宕機(jī)的事情，kafka controller也會(huì)通知zookeeper，zookeeper就會(huì)通知其他的kafka broker。

二、Kafka架構(gòu)

三、Kafka中的術(shù)語解釋概述

Broker【服務(wù)器節(jié)點(diǎn)】

Kafka 集群包含一個(gè)或多個(gè)服務(wù)器，服務(wù)器節(jié)點(diǎn)稱為broker。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ù)不均衡。

Topic【主題】

每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為Topic。（物理上不同Topic的消息分開存儲(chǔ)，邏輯上一個(gè)Topic的消息雖然保存于
一個(gè)或多個(gè)broker上但用戶只需指定消息的Topic即可生產(chǎn)或消費(fèi)數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）。類似于數(shù)據(jù)庫的表名。在每個(gè)broker上都
可以創(chuàng)建多個(gè)topic。

Partition【分區(qū)】

• topic中的數(shù)據(jù)分割為一個(gè)或多個(gè)partition。每個(gè)topic至少有一個(gè)partition。topic的數(shù)據(jù)數(shù)據(jù)會(huì)寫入到不同的partition。

• 每個(gè)Partition在物理上對(duì)應(yīng)一個(gè)文件夾，該文件夾下存儲(chǔ)這個(gè)Partition的所有消息和索引文件。

• partition中的數(shù)據(jù)是有序的，不同partition間的數(shù)據(jù)丟失了數(shù)據(jù)的順序。

• 如果topic有多個(gè)partition，消費(fèi)數(shù)據(jù)時(shí)就不能保證數(shù)據(jù)的順序。

• 在需要嚴(yán)格保證消息的消費(fèi)順序的場(chǎng)景下，需要將partition數(shù)目設(shè)為1。

上面說到數(shù)據(jù)會(huì)寫入到不同的分區(qū)，那kafka為什么要做分區(qū)呢？相信大家應(yīng)該也能猜到，分區(qū)的主要目的是：

1. 方便擴(kuò)展。因?yàn)橐粋€(gè)topic可以有多個(gè)partition，所以我們可以通過擴(kuò)展機(jī)器去輕松的應(yīng)對(duì)日益增長(zhǎng)的數(shù)據(jù)量。

2. 提高并發(fā)。以partition為讀寫單位，可以多個(gè)消費(fèi)者同時(shí)消費(fèi)數(shù)據(jù)，提高了消息的處理效率。

熟悉負(fù)載均衡的朋友應(yīng)該知道，當(dāng)我們向某個(gè)服務(wù)器發(fā)送請(qǐng)求的時(shí)候，服務(wù)端可能會(huì)對(duì)請(qǐng)求做一個(gè)負(fù)載，將流量分發(fā)到不同的服務(wù)器，那在kafka中，如果某個(gè)topic有多個(gè)partition，producer又怎么知道該將數(shù)據(jù)發(fā)往哪個(gè)partition呢？

kafka中有幾個(gè)原則：

1. partition在寫入的時(shí)候可以指定需要寫入的partition，如果有指定，則寫入對(duì)應(yīng)的partition。

2. 如果沒有指定partition，但是設(shè)置了數(shù)據(jù)的key，則會(huì)根據(jù)key的值hash出一個(gè)partition。

3. 如果既沒指定partition，又沒有設(shè)置key，則會(huì)輪詢選出一個(gè)partition。

保證消息不丟失是一個(gè)消息隊(duì)列中間件的基本保證，那producer在向kafka寫入消息的時(shí)候，怎么保證消息不丟失呢？

那就是通過ACK應(yīng)答機(jī)制！在生產(chǎn)者向隊(duì)列寫入數(shù)據(jù)的時(shí)候可以設(shè)置參數(shù)來確定是否確認(rèn)kafka接收到數(shù)據(jù)，這個(gè)參數(shù)可設(shè)置的值為0、1、all。

• 0代表producer往集群發(fā)送數(shù)據(jù)不需要等到集群的返回，不確保消息發(fā)送成功。安全性最低但是效率最高。

• 1代表producer往集群發(fā)送數(shù)據(jù)只要leader應(yīng)答就可以發(fā)送下一條，只確保leader發(fā)送成功。

• all代表producer往集群發(fā)送數(shù)據(jù)需要所有的follower都完成從leader的同步才會(huì)發(fā)送下一條，確保leader發(fā)送成功和所有的副本都完成備份。安全性最高，但是效率最低。

最后要注意的是，如果往不存在的topic寫數(shù)據(jù)，能不能寫入成功呢？kafka會(huì)自動(dòng)創(chuàng)建topic，分區(qū)和副本的數(shù)量根據(jù)默認(rèn)配置都是1。

Producer【生產(chǎn)者】

生產(chǎn)者即數(shù)據(jù)的發(fā)布者，該角色將消息發(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。

Consumer【消費(fèi)者】

消費(fèi)者可以從broker中讀取數(shù)據(jù)。消費(fèi)者可以消費(fèi)多個(gè)topic中的數(shù)據(jù)。

Consumer Group【消費(fèi)者組】

每個(gè)Consumer屬于一個(gè)特定的Consumer Group（可為每個(gè)Consumer指定group name，若不指定group name則屬于默認(rèn)的group）。

同一個(gè)topic下的每個(gè)partition中message只能被組（Consumer group ）中的一個(gè)consumer消費(fèi)，如果想讓一個(gè)message可以被多個(gè)consumer消費(fèi)的話，那么這些consumer必須在不同的Consumer group。所以如果想同時(shí)對(duì)一個(gè)topic做消費(fèi)的話，啟動(dòng)多個(gè)consumer group就可以了，但是要注意的是，這里的多個(gè)consumer的消費(fèi)都必須是順序讀取partition里面的message，新啟動(dòng)的consumer默認(rèn)從partition隊(duì)列最頭端最新的地方開始阻塞的讀message。它不能像AMQ那樣可以多個(gè)BET作為consumer去互斥的（for update悲觀鎖）并發(fā)處理message，這是因?yàn)槎鄠€(gè)BET去消費(fèi)一個(gè)Queue中的數(shù)據(jù)的時(shí)候，由于要保證不能多個(gè)線程拿同一條message，所以就需要行級(jí)別悲觀所（for update）,這就導(dǎo)致了consume的性能下降，吞吐量不夠。而kafka為了保證吞吐量，只允許同一個(gè)consumer group下的一個(gè)consumer線程去訪問一個(gè)partition。如果覺得效率不高的時(shí)候，可以加partition的數(shù)量來橫向擴(kuò)展，那么再加新的consumer thread去消費(fèi)。如果想多個(gè)不同的業(yè)務(wù)都需要這個(gè)topic的數(shù)據(jù)，起多個(gè)consumer group就好了，大家都是順序的讀取message，offsite的值互不影響。這樣沒有鎖競(jìng)爭(zhēng)，充分發(fā)揮了橫向的擴(kuò)展性，吞吐量極高。這也就形成了分布式消費(fèi)的概念。

當(dāng)啟動(dòng)一個(gè)consumer group去消費(fèi)一個(gè)topic的時(shí)候，無論topic里面有多少個(gè)partition，無論我們consumer group里面配置了多少個(gè)consumer thread，這個(gè)consumer group下面的所有consumer thread一定會(huì)消費(fèi)全部的partition；即便這個(gè)consumer group下只有一個(gè)consumer thread，那么這個(gè)consumer thread也會(huì)去消費(fèi)所有的partition。因此，最優(yōu)的設(shè)計(jì)就是，consumer group下的consumer thread的數(shù)量等于partition數(shù)量，這樣效率是最高的。

• 當(dāng)consumer group里面的consumer數(shù)量小于這個(gè)topic下的partition數(shù)量的時(shí)候，就會(huì)出現(xiàn)一個(gè)conusmer thread消費(fèi)多個(gè)partition的情況，總之是這個(gè)topic下的partition都會(huì)被消費(fèi)。
• 如果consumer group里面的consumer數(shù)量等于這個(gè)topic下的partition數(shù)量的時(shí)候，此時(shí)效率是最高的，每個(gè)partition都有一個(gè)consumer thread去消費(fèi)。
• 當(dāng)consumer group里面的consumer數(shù)量大于這個(gè)topic下的partition數(shù)量的時(shí)候，就會(huì)有consumer thread空閑。

多個(gè)Consumer Group下的consumer可以消費(fèi)同一條message，但是這種消費(fèi)也是以o（1）的方式順序的讀取message去消費(fèi),，所以一定會(huì)重復(fù)消費(fèi)這批message的，不能向AMQ那樣多個(gè)BET作為consumer消費(fèi)（對(duì)message加鎖，消費(fèi)的時(shí)候不能重復(fù)消費(fèi)message）

Leader【領(lǐng)導(dǎo)者】

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

Follower【跟隨者】

• Follower跟隨Leader，所有寫請(qǐng)求都通過Leader路由，數(shù)據(jù)變更會(huì)廣播給所有Follower，Follower與Leader保持?jǐn)?shù)據(jù)同步。
• 如果Leader失效，則從Follower中選舉出一個(gè)新的Leader。
• 當(dāng)Follower與Leader掛掉、卡住或者同步太慢，leader會(huì)把這個(gè)follower從“in sync replicas”（ISR）列表中刪除，重新創(chuàng)建一個(gè)Follower。

Replica【副本】

每個(gè)partition可以在其他的kafka broker節(jié)點(diǎn)上存副本，以便某個(gè)kafka broker節(jié)點(diǎn)宕機(jī)不會(huì)影響這個(gè)kafka集群。
存replica副本的方式是按照kafka broker的順序存。
例如有5個(gè)kafka broker節(jié)點(diǎn)，某個(gè)topic有3個(gè)partition，每個(gè)partition存2個(gè)副本，那么partition1存broker1,broker2，partition2存broker2,broker3。。。以此類推（replica副本數(shù)目不能大于kafka broker節(jié)點(diǎn)的數(shù)目，否則報(bào)錯(cuò)。這里的replica數(shù)其實(shí)就是partition的副本總數(shù)，其中包括一個(gè)leader，其他的就是copy副本）。這樣如果某個(gè)broker宕機(jī)，其實(shí)整個(gè)kafka內(nèi)數(shù)據(jù)依然是完整的。但是，replica副本數(shù)越高，系統(tǒng)雖然越穩(wěn)定，但是會(huì)帶來資源和性能上的下降；replica副本少的話，也會(huì)造成系統(tǒng)丟數(shù)據(jù)的風(fēng)險(xiǎn)。

• 傳送消息：producer先把message發(fā)送到partition leader，再由leader發(fā)送給其他partition follower（如果讓producer發(fā)送給每個(gè)replica那就太慢了）。 再向Producer發(fā)送ACK前需要保證有多少個(gè)Replica已經(jīng)收到該消息：根據(jù)ack配的個(gè)數(shù)而定。

• 處理某個(gè)Replica不工作的情況：如果這個(gè)部工作的partition replica不在ack列表中，就是producer在發(fā)送消息到partition leader上，partition leader向partition follower發(fā)送message沒有響應(yīng)而已，這個(gè)不會(huì)影響整個(gè)系統(tǒng)，也不會(huì)有什么問題。如果這個(gè)不工作的partition replica在ack列表中的話，producer發(fā)送的message的時(shí)候會(huì)等待這個(gè)不工作的partition replca寫message成功，但是會(huì)等到time out，然后返回失敗因?yàn)槟硞€(gè)ack列表中的partition replica沒有響應(yīng)，此時(shí)kafka會(huì)自動(dòng)的把這個(gè)部工作的partition replica從ack列表中移除，以后的producer發(fā)送message的時(shí)候就不會(huì)有這個(gè)ack列表下的這個(gè)部工作的partition replica了。

• 處理Failed Replica恢復(fù)回來的情況：如果這個(gè)partition replica之前不在ack列表中，那么啟動(dòng)后重新受Zookeeper管理即可，之后producer發(fā)送message的時(shí)候，partition leader會(huì)繼續(xù)發(fā)送message到這個(gè)partition follower上。如果這個(gè)partition replica之前在ack列表中，此時(shí)重啟后，需要把這個(gè)partition replica再手動(dòng)加到ack列表中。（ack列表是手動(dòng)添加的，出現(xiàn)某個(gè)部工作的partition replica的時(shí)候自動(dòng)從ack列表中移除的）。

四、Kafka可視化管理工具

【Kafka可視化工具】kafka-manager
kafka-manager安裝及基本使用

【Kafka可視化工具】Offset Explorer
Kafka-Offset Explorer安裝及基本使用