一、Kafka簡介

Kafka是什么

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

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

1.1 Kafka的特性:

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

• 可擴展性：kafka集群支持熱擴展

• 持久性、可靠性：消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失

• 容錯性：允許集群中節(jié)點失敗（若副本數(shù)量為n,則允許n-1個節(jié)點失敗）

• 高并發(fā)：支持?jǐn)?shù)千個客戶端同時讀寫

1.2 Kafka的使用場景：

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

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

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

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

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

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

• 事件源

1.3 Kakfa的設(shè)計思想

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

二、Kafka架構(gòu)

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

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

Kafka 集群包含一個或多個服務(wù)器，服務(wù)器節(jié)點稱為broker。broker存儲topic的數(shù)據(jù)。

• 如果某topic有N個partition，集群有N個broker，那么每個broker存儲該topic的一個partition。

• 如果某topic有N個partition，集群有(N+M)個broker，那么其中有N個broker存儲該topic的一個partition，剩下的M個broker不存儲該topic的partition數(shù)據(jù)。

• 如果某topic有N個partition，集群中broker數(shù)目少于N個，那么一個broker存儲該topic的一個或多個partition。在實際生產(chǎn)環(huán)境中，盡量避免這種情況的發(fā)生，這種情況容易導(dǎo)致Kafka集群數(shù)據(jù)不均衡。

Topic【主題】

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

Partition【分區(qū)】

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

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

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

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

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

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

1. 方便擴展。因為一個topic可以有多個partition，所以我們可以通過擴展機器去輕松的應(yīng)對日益增長的數(shù)據(jù)量。

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

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

kafka中有幾個原則：

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

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

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

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

那就是通過ACK應(yīng)答機制！在生產(chǎn)者向隊列寫入數(shù)據(jù)的時候可以設(shè)置參數(shù)來確定是否確認(rèn)kafka接收到數(shù)據(jù)，這個參數(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的同步才會發(fā)送下一條，確保leader發(fā)送成功和所有的副本都完成備份。安全性最高，但是效率最低。

最后要注意的是，如果往不存在的topic寫數(shù)據(jù)，能不能寫入成功呢？kafka會自動創(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ā)送的消息，存儲到一個partition中，生產(chǎn)者也可以指定數(shù)據(jù)存儲的partition。

Consumer【消費者】

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

Consumer Group【消費者組】

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

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

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

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

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

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

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

Follower【跟隨者】

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

Replica【副本】

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

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

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

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

四、Kafka可視化管理工具

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

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