kafka背景及架構(gòu)介紹

1、Kafka背景及架構(gòu)介紹目錄1、使用消息系統(tǒng)的優(yōu)點2、常用Message Queue介紹和對比3、選擇Kafka的原因4、Kafka的架構(gòu)及相關(guān)概念介紹5、Kafka啟動和監(jiān)控6、Kafka日志文件存儲及生產(chǎn)、消費的原理7、Kafka穩(wěn)定性、容錯性、異常情況的介紹8、Kafka工具的使用9、Kafka項目中的應(yīng)用為何使用消息系統(tǒng)1、數(shù)據(jù)持久化，數(shù)據(jù)緩沖，異步通信。保證數(shù)據(jù)不丟失，通過緩沖層來幫助任務(wù)最高效率的執(zhí)行，可存儲大量歷史數(shù)據(jù)可以異步處理。2、擴展性，解耦處理過程。增大消息入隊和處理的頻率是很容易的，只要另外增加處理過程即可。不需要改變代碼、不需要調(diào)節(jié)參數(shù)。3、靈活性，峰值處理能力。不會

2、因為突發(fā)的超負荷的請求而完全崩潰。4、可恢復(fù)性。系統(tǒng)的一部分組件失效時，不會影響到整個系統(tǒng)。加入隊列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。5、時序性。在大多使用場景下，數(shù)據(jù)處理的順序都很重要。常用Message Queue對比對比RabbitMQRocketMQkafka模式發(fā)布訂閱發(fā)布訂閱發(fā)布訂閱持久化支持（內(nèi)存/硬盤）支持（磁盤）支持（磁盤）事務(wù)支持支持不支持集群支持支持支持負載均衡支持支持支持堆積能力磁盤容量+內(nèi)存磁盤容量磁盤容量（水平擴展，不同broker可以存儲在不同磁盤上）消息丟失不會丟失理論上不會丟失理論上不會丟失消息重復(fù)可以使用消息確認等機制控制-理論上會有重復(fù)消息失敗重試機制消

3、費失敗支持定時重試消費失敗支持定時重試，每次重試間隔時間順延Kafka消費失敗不支持重試客戶端支持語言Java、C、C+、Python、PHP、Perl等Java、C+（不成熟）Java、C+等最大吞吐量時cpu、內(nèi)存占用cpu占用非常高，內(nèi)存占用根據(jù)配置（配置的太低對性能影響很大）-cpu占用不高，每個broker大概10%，內(nèi)存1G左右Producer吞吐量1.8w/s(4個broker)，消費者消費對其吞吐量影響很大-13w/s(4個broker，11個分區(qū))，消費者消費對其吞吐量影響很小Consumer吞吐量6千/s（3個消費者），峰值數(shù)據(jù)非常容易積壓，一旦積壓消費速度有很大影響，消費

4、者增加速度成倍下降-2.7w/s（3個消費者，11個分區(qū)），能累積很多數(shù)據(jù)，累積不會對消費速度有影響，增加消費者對消費速度的影響沒有rabbitmq那么大。測試環(huán)境（203）：Model: Dell Vostro 3800 CPU: Intel(R) Core(TM) i3-4170 CPU 3.70GHzRAM: 16 GbOS: CentOS 7.3.1611 x86_64rabbitmq1、RabbitMQ是使用Erlang語言開發(fā)的開源消息隊列系統(tǒng)，基于AMQP協(xié)議來實現(xiàn)。2、AMQP的主要特征是面向消息、隊列、路由（包括點對點和發(fā)布/訂閱）、可靠性、安全。3、AMQP協(xié)議更多用在企業(yè)

5、系統(tǒng)內(nèi)，對數(shù)據(jù)一致性、穩(wěn)定性和可靠性要求很高的場景，對性能和吞吐量的要求還在其次。RocketMQ結(jié)合部署結(jié)構(gòu)圖，描述集群工作流程：1、啟動Namesrv，Namesrv起來后監(jiān)聽端口，等待Broker、Produer、Consumer連上來，相當(dāng)于一個路由控制中心。2、Broker啟動，跟所有的Namesrv保持長連接，定時發(fā)送心跳包。心跳包中包含當(dāng)前Broker信息(IP+端口等)以及存儲所有topic信息。注冊成功后，namesrv集群中就有Topic跟Broker的映射關(guān)系。3、收發(fā)消息前，先創(chuàng)建topic，創(chuàng)建topic時需要指定該topic要存儲在哪些Broker上。也可以在發(fā)送消

6、息時自動創(chuàng)建Topic。4、Producer發(fā)送消息，啟動時先跟Namesrv集群中的其中一臺建立長連接，并從Namesrv中獲取當(dāng)前發(fā)送的Topic存在哪些Broker上，然后跟對應(yīng)的Broker建長連接，直接向Broker發(fā)消息。5、Consumer跟Producer類似。跟其中一臺Namesrv建立長連接，獲取當(dāng)前訂閱Topic存在哪些Broker，然后直接跟Broker建立連接通道，開始消費消息。Kafka架構(gòu)1、Kafka是Scala和Java編寫的分布式發(fā)布-訂閱消息系統(tǒng)。2、Kafka主要特點是基于Pull的模式來處理消息消費，追求高吞吐量，采用消息的批量處理，zero-copy

7、機制，數(shù)據(jù)的存儲和獲取是本地磁盤順序批量操作，具有O(1)的復(fù)雜度。3、Kafka對消息的重復(fù)、丟失、錯誤沒有嚴(yán)格要求，需要開發(fā)者增加控制機制。1、KAFKA比較成熟，并且以時間復(fù)雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數(shù)據(jù)也能保證常數(shù)時間復(fù)雜度的訪問性能。2、高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條以上消息的傳輸。順序?qū)懘疟P（順序?qū)懘疟P效率比隨機寫內(nèi)存要高，保障 kafka 吞吐率。3、支持Kafka Server間的消息分區(qū)，及分布式消費，同時保證每個Partition內(nèi)的消息順序傳輸，采用消息的批量處理，zero-copy機制。4、同時支持離線數(shù)

8、據(jù)處理和實時數(shù)據(jù)處理。5、支持在線水平擴展，隨時增加zookeeper節(jié)點，每個broker還可以分布在不同磁盤，增加處理性能。為何使用KAFKAKafka相關(guān)概念介紹zookeeper集群Kafka系統(tǒng)強依賴的組件。其存儲了Kafka核心原數(shù)據(jù) (如topic信息配置、broker信息、 消費分組等等，相當(dāng)于DB充當(dāng)了Kafka的配置管理中心) 。 Kafka的leader選舉(如coordinator選舉、controller選舉、partition leader選舉等等)，同樣也會借助于zookeeper。Broker消息緩存代理，Kafka集群包含一個或多個服務(wù)器，這些服務(wù)器被稱為Br

9、oker，負責(zé)消息的存儲于轉(zhuǎn)發(fā)，作為代理對外提供生產(chǎn)和消費服務(wù)。Topic消息主題(類別)，邏輯上的概念，特指Kafka處理的消息源的不同分類，用戶可以根據(jù)自己的業(yè)務(wù)形態(tài)將不同業(yè)務(wù)類別的消息分別存儲到不同Topic。用戶生產(chǎn)和消費時只需指定所關(guān)注的topic即可，不用關(guān)注該topic的數(shù)據(jù)存放的具體位置。PartitionTopic物理上的分組，在創(chuàng)建Topic時可以指定分區(qū)的數(shù)量，每個partition是一個有序的隊列，按生產(chǎn)順序存儲著每條消息，而且每條消息都會分配一個64bit的自增長的有序offset(相當(dāng)于消息id)。Partition是整個Kafka可以平行擴展的關(guān)鍵因素。Repli

10、cation副本，topic級別的配置，可以理解為topic消息的副本數(shù)。Kafka 0.8版本加入的概念，主要目的就是提高系統(tǒng)的可用性。防止broker意外崩潰導(dǎo)致部分partition不可以服務(wù)。Leaderreplica 中的一個角色， producer 和 consumer 只跟 leader 交互。 Followerreplica 中的一個角色，從 leader 中復(fù)制數(shù)據(jù)。ISRIn-Sync Replicas ，Kafka用來維護跟上leader數(shù)據(jù)的broker列表,當(dāng)leader崩潰后，優(yōu)先從該列中選舉leader。ProducerProducer 生產(chǎn)者，采用Push方式進

11、行消息發(fā)布生產(chǎn)。 Producer可以通過與zookeeper連接獲取broker信息, topic信息等等元數(shù)據(jù)，然后再與broker交互進行消息發(fā)布。Consumer消費者，采用Pull方式，從Broker端拉取消息并進行處理。當(dāng)采用訂閱方式訂閱感興趣的Topic時，Consumer必須屬于一個消費分組，而且Kafka保證同一個Topic的一條消息只能被同一個消費分組（GroupId）中的一個Consumer消費，但多個消費分組可以同時消費這一條消息。Kafka啟動和監(jiān)控管理1、Zookeeper啟動 cd /root/kafka/kafka-src/bin zkServer.sh sta

12、rt /root/kafka/kafka-src/config/perties zkServer.sh start /root/kafka/kafka-src/config/perties2、Kafka的brokers啟動 cd /root/kafka/kafka-src/bin cat config/perties kafka-server-start.sh -daemon /root/kafka/kafka-src/config/perties & kafka-server-start1.sh -daemon /root/kafka/kafka-src/config/perties & k

13、afka-server-start2.sh -daemon /root/kafka/kafka-src/config/perties & kafka-server-start3.sh -daemon /root/kafka/kafka-src/config/perties &3、創(chuàng)建topic kafka-topics.sh -create -zookeeper 32:2182,32:2183 -replication-factor 2 -partitions 11 -topic mapgoo.gps4、啟動監(jiān)控 cd /rootkafka/kafka-manager-3 cat conf/a

14、pplication.conf bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9001 &5、登錄監(jiān)控 網(wǎng)頁輸入03:9001broker.id=0listeners=PLAINTEXT:/32:9092log.dirs=/opt/kafka/kafka-src/bin/kafka-logszookeeper.connect=:2182,:2183kafka-manager.zkhosts=:2182, :2183web監(jiān)控的zookeepers和brokers的展示web監(jiān)控的topic和partition

15、的展示Kafka日志文件的存儲同一個topic下有多個不同的partition，每個partition為一個目錄，partition命名的規(guī)則是topic的名稱加上一個序號，序號從0開始。 每一個partition目錄下的文件被平均切割成大小相等的數(shù)據(jù)文件，和一一對應(yīng)的索引文件。每一個數(shù)據(jù)文件都被稱為一個段（segment file），但每個段消息數(shù)量不一定相等，這種特性能夠使得老的segment可以被快速清除。默認保留7天的數(shù)據(jù)。Segment文件命名的規(guī)則：partition全局的第一個segment從0（20個0）開始，后續(xù)的每一個segment文件名是上一個segment文件中最后一條

16、消息的offset值。那么這樣命令有什么好處呢？假如有個消費者已經(jīng)消費到了368776（offset值為368776），那么現(xiàn)在要繼續(xù)消費的話，怎么做？分2個步驟，第1步是從所有文件log文件的的文件名中找到對應(yīng)的log文件，第368776條數(shù)據(jù)位于上圖中的“00000000000000368769.log”這個文件中，這一步涉及到一個常用的算法叫做“二分查找法”（假如我現(xiàn)在給你一個offset值讓你去找，你首先是將所有的log的文件名進行排序，然后通過二分查找法進行查找，很快就能定位到某一個文件，緊接著拿著這個offset值到其索引文件中找這條數(shù)據(jù)究竟存在哪里）；第2步是到index文件中去

17、找第368776條數(shù)據(jù)所在的位置。index文件中并沒有為數(shù)據(jù)文件中的每條消息都建立索引，而是采用了稀疏存儲的方式，每隔一定字節(jié)的數(shù)據(jù)建立一條索引。Kafka Replication的數(shù)據(jù)流producer 發(fā)布消息的過程Producer在發(fā)布消息到某個Partition時，先通過ZooKeeper找到該Partition的Leader，然后無論該Topic的Replication Factor為多少（也即該Partition有多少個Replica），Producer只將該消息發(fā)送到該Partition的Leader。Leader會將該消息寫入其本地Log。每個Follower都從Leader

18、 pull數(shù)據(jù)。這種方式上，F(xiàn)ollower存儲的數(shù)據(jù)順序與Leader保持一致。Follower在收到該消息并寫入其Log后，向Leader發(fā)送ACK。一旦Leader收到了ISR中的所有Replica的ACK，該消息就被認為已經(jīng)commit了，Leader將增加HW（俗稱高水位）并且向Producer發(fā)送ACK。kafka副本同步隊列（ISR）副本數(shù)對Kafka的吞吐率是有一定的影響，但極大的增強了可用性。默認情況下Kafka的replica數(shù)量為1，即每個partition都有一個唯一的leader，為了確保消息的可靠性，通常應(yīng)用中將其值小設(shè)置為大于1，比如3。 所有的副本（replic

19、as）統(tǒng)稱為Assigned Replicas，即AR。ISR是AR中的一個子集，由leader維護ISR列表，follower從leader同步數(shù)據(jù)有一些延遲。kafka數(shù)據(jù)可靠性和持久性保證當(dāng)producer向leader發(fā)送數(shù)據(jù)時，可以通過request.required.acks參數(shù)來設(shè)置數(shù)據(jù)可靠性的級別：1（默認）：這意味著producer在ISR中的leader已成功收到的數(shù)據(jù)并得到確認后發(fā)送下一條message。如果leader宕機了，則會丟失數(shù)據(jù)。0：這意味著producer無需等待來自broker的確認而繼續(xù)發(fā)送下一批消息。這種情況下數(shù)據(jù)傳輸效率最高，但是數(shù)據(jù)可靠性確是最低的

20、。-1：producer需要等待ISR中的所有follower都確認接收到數(shù)據(jù)后才算一次發(fā)送完成，可靠性最高。但是這樣也不能保證數(shù)據(jù)不丟失，比如當(dāng)ISR中只有l(wèi)eader時（前面ISR那一節(jié)講到，ISR中的成員由于某些情況會增加也會減少，最少就只剩一個leader），這樣就變成了acks=1的情況。接下來對acks=1和-1的兩種情況進行詳細分析：request.required.acks=1 producer發(fā)送數(shù)據(jù)到leader，leader寫本地日志成功，返回客戶端成功；此時ISR中的副本還沒有來得及拉取該消息，leader就宕機了，那么此次發(fā)送的消息就會丟失。注：LEO（log end

21、 offset）。request.required.acks=-1同步（Kafka默認為同步，即producer.type=sync）的發(fā)送模式，replication.factor=2且min.insync.replicas=2的情況下，不會丟失數(shù)據(jù)。有兩種典型情況。acks=-1的情況下，數(shù)據(jù)發(fā)送到leader, ISR的follower全部完成數(shù)據(jù)同步后，leader此時掛掉，那么會選舉出新的leader，數(shù)據(jù)不會丟失。acks=-1的情況下，數(shù)據(jù)發(fā)送到leader后 ，部分ISR的副本同步，leader此時掛掉。比如follower1h和follower2都有可能變成新的leader,

22、 producer端會得到返回異常，producer端會重新發(fā)送數(shù)據(jù)，follower1被選舉為新的leader的話，數(shù)據(jù)可能會重復(fù)。當(dāng)然如果在leader crash的時候，follower2還沒有同步到任何數(shù)據(jù)，而且follower2被選舉為新的leader的話，這樣消息就不會重復(fù)。上圖中，還有另一種情況，如果在Leader掛掉的時候，follower1同步了消息4,5，follower2同步了消息4，與此同時follower2被選舉為leader，那么此時follower1中的多出的消息5該做如何處理呢？ 這里就需要HW和LEO的協(xié)同配合了。如前所述，一個partition中的ISR列表

23、中，leader的HW是所有ISR列表里副本中最小的那個的LEO。類似于木桶原理，水位取決于最低那塊短板。下面看看Leader掛掉，新的Leader產(chǎn)生的現(xiàn)象。主題iteblog的分布如下：手動停止部分Broker1/2/4，Partition 0的Leader由原來的1變?yōu)?，Partition 1的Leader由原來的2變?yōu)?，Partition 2的Leader由原來的3變?yōu)?，Partition 3的Leader由原來的4變?yōu)?。主題iteblog的分布如下：consumer 消費消息kafka 提供了兩套 consumer API：The high-level consumer AP

24、I high-level consumer API 提供了 consumer group 的語義，一個消息只能被 group 內(nèi)的一個 consumer 所消費，且consumer 消費消息時不關(guān)注 offset，最后一個 offset 由 zookeeper 保存。該接口支持負載均衡，當(dāng)有 consumer 加入或退出、以及 partition 的改變（如 broker 加入或退出）時會觸發(fā) rebalance，即重新負載所有的consumer。The low-level consumer API 使用Low Level Consumer (Simple Consumer)的主要原因是，用戶

25、希望比Consumer Group更好的控制數(shù)據(jù)的消費，吞吐量可以達到更高，比如： 同一條消息讀多次； 只讀取某個Topic的部分Partition； 管理事務(wù)，從而確保每條消息被處理一次，且僅被處理一次。與Consumer Group相比，Low Level Consumer要求用戶做大量的額外工作。 必須在應(yīng)用程序中跟蹤offset，從而確定下一條應(yīng)該消費哪條消息（可以使用redis或者文件存儲offset） 應(yīng)用程序需要通過程序獲知每個Partition的Leader是誰 必須處理Leader的變化 必須自己開發(fā)消費者的監(jiān)控（可以定時查詢redis 存儲的偏移量來監(jiān)控）使用Low Lev

26、el Consumer的一般流程如下 查找到一個“活著”的Broker，并且找出每個Partition的Leader 找出每個Partition的Follower 定義好請求，該請求應(yīng)該能描述應(yīng)用程序需要哪些數(shù)據(jù) Fetch數(shù)據(jù) 識別Leader的變化，并對之作出必要的響應(yīng)kafka 在 zookeeper 中的存儲結(jié)構(gòu)Kafka創(chuàng)建topicKafka創(chuàng)建topic命令很簡單，一條命令足矣：bin/kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 3 -partitions 8 -topic itebl

27、og命令行部分 我們發(fā)起topic創(chuàng)建命令之后，Kafka會做一些基本的校驗，比如是否同時指定了分區(qū)數(shù)、副本因子或是topic名字中是否含有非法字符等。 做完基本的校驗之后，Kafka會從zookeeper的/brokers/ids下獲取集群當(dāng)前存活broker列表然后開始執(zhí)行副本的分配工作。首先，分區(qū)副本的分配有以下2個目標(biāo)： 盡可能地在各個broker之間均勻地分配副本； 如果分區(qū)的某個副本被分配到了一個broker，那么要盡可能地讓該分區(qū)的其他副本均勻地分配到其他broker上；后臺邏輯部分 所謂的后臺邏輯其實是由Kafka的controller負責(zé)提供的。Kafka的controlle

28、r內(nèi)部保存了很多信息，其中有一個分區(qū)狀態(tài)機，用于記錄topic各個分區(qū)的狀態(tài)。這個狀態(tài)機內(nèi)部注冊了一些zookeeper監(jiān)聽器。監(jiān)聽器一方面會更新controller的緩存信息(比如更新集群當(dāng)前所有的topic列表以及更新新增topic的分區(qū)副本分配方案緩存等)，另一方面就是創(chuàng)建對應(yīng)的分區(qū)及其副本對象并為每個分區(qū)確定leader副本及ISR。有了Replication機制后，每個Partition可能有多個備份。某個Partition的Replica列表叫作AR（Assigned Replicas），AR中的第一個Replica即為“Preferred Replica”。創(chuàng)建一個新的Topic

29、或者給已有Topic增加Partition時，Kafka保證Preferred Replica被均勻分布到集群中的所有Broker上。理想情況下，Preferred Replica會被選為Leader。以上兩點保證了所有Partition的Leader被均勻分布到了集群當(dāng)中，這一點非常重要，因為所有的讀寫操作都由Leader完成，若Leader分布過于集中，會造成集群負載不均衡。但是，隨著集群的運行，該平衡可能會因為Broker的宕機而被打破，該工具就是用來幫助恢復(fù)Leader分配的平衡。假設(shè)有0-7個Broker，Broker 1/2/4都被停止， Replica列表分布如下：Preferr

30、edReplica Leader Election Tool再啟動Broker 1后， Broker 1 不會再成為Leader ，這時負載不均勻， Isr變化如下：再啟動Broker 1后， 運行該工具后， Leader變化如下：該工具的設(shè)計目標(biāo)與Preferred Replica Leader Election Tool有些類似，都旨在促進Kafka集群的負載均衡。不同的是，Preferred Replica Leader Election只能在Partition的AR范圍內(nèi)調(diào)整其Leader，使Leader分布均勻，而該工具還可以調(diào)整Partition的AR。 該工具有三種使用模式 （1

31、）generate模式，給定需要重新分配的Topic，自動生成reassign plan（并不執(zhí)行） $KAFKA_HOME/bin/kafka-reassign-partitions.sh -zookeeper :2181 -topics-to-move-json-file /tmp/topics-to-move.json -broker-list 2,3 -generate （2）execute模式，根據(jù)指定的reassign plan重新分配Partition $KAFKA_HOME/bin/kafka-reassign-partitions.sh -zookeeper :2181 -r

32、eassignment-json-file /tmp/reassign-plan.json -execute （3）verify模式，驗證重新分配Partition是否成功 $KAFKA_HOME/bin/kafka-reassign-partitions.sh -zookeeper :2181 -reassignment-json-file /tmp/reassign-plan.json verify 接下來用Topic Tool再次驗證： bin/kafka-topics.sh -zookeeper :2181 -describe -topic iteblogKafka Reassign

33、Partitions Tool該工具旨在從整個集群的Broker上收集狀態(tài)改變?nèi)罩荆⑸梢粋€集中的格式化的日志以幫助診斷狀態(tài)改變相關(guān)的故障。每個Broker都會將其收到的狀態(tài)改變相關(guān)的的指令存于名為state-change.log的日志文件中。某些情況下，Partition的Leader election可能會出現(xiàn)問題，此時我們需要對整個集群的狀態(tài)改變有個全局的了解從而診斷故障并解決問題。該工具將集群中相關(guān)的state-change.log日志按時間順序合并，同時支持用戶輸入時間范圍和目標(biāo)Topic及Partition作為過濾條件，最終將格式化的結(jié)果輸出。 用法 bin/kafka-run-

34、class.sh kafka.tools.StateChangeLogMerger -logs /opt/kafka_2.11-/logs/state-change.log -topic iteblog -partitions 0,1,2,3,4,5,6,7State Change Log Merge ToolKafka項目中的部署框架MRS的數(shù)據(jù)按objectid對最大分區(qū)ID進行取模push到不同的partition； MRS可以多線程或者多進程push。SAS、VTS和RNS必須使用不同的組ID（GroupId），這樣才能拉取全量的同分數(shù)據(jù)； 比如SAS設(shè)置GroupId為0，VTS設(shè)置

35、GroupId為1 ， RNS設(shè)置GroupId為2，在同一個GroupId中，每個partition需對應(yīng)一個線程（或者一個進程），或者一個線程輪詢讀取多個partition（這種方法會影響吞吐量），如果多個線程消費同一個partition，那么有的線程永遠讀不到數(shù)據(jù)，會造成線程浪費。 使用librdkafka時，一個進程消費partition不宜過多，主要還是librdkafka給每個partition創(chuàng)建多個線程， partition越多創(chuàng)建的線程越多，比如，一個線程消費51個分區(qū)，創(chuàng)建的線程數(shù)有300個左右。 Producer的接口 CProducerKafka生產(chǎn)者 類方法：int init(const std:string& brokers, const std:string& groupid, const std:string& topicname, const int maxpartition, CMsgDeliveredCb* pMsgDelivered = NULL); templa