kafka設(shè)計(jì)原理分析集群搭建與使用

KafkaKafkaScala對(duì)于熟悉JMS（JavaMessageService）規(guī)范的同學(xué)來(lái)說(shuō)，消息系統(tǒng)已經(jīng)不是什么新概念了(例如ActiveMQ，RabbitMQ等)。Kafka擁有作為一個(gè)消息系統(tǒng)應(yīng)該具備的功能，但是確有著獨(dú)特的設(shè)計(jì)。可以這樣來(lái)說(shuō)，Kafka借鑒了JMS規(guī)范的思想，但是確并沒(méi)有完全遵循JMS規(guī)范。kafka是一個(gè)分布式的，分區(qū)的消息(稱(chēng)之為commitlog)服務(wù)。它提供一個(gè)消Topic:Kafka按照Topic分類(lèi)來(lái)消Producer：發(fā)布(publish)消息到Topic的進(jìn)程稱(chēng)之為生產(chǎn)者Consumer：訂閱(subscribe)Topic并且處理Topic中消息的進(jìn)程稱(chēng)之BrokerKafka以集群的方式運(yùn)行，集群中的每一臺(tái)服務(wù)器稱(chēng)之為一個(gè)因此，從一個(gè)較高的層面上來(lái)看，producers通過(guò)網(wǎng)絡(luò)發(fā)送消息到Kafka集群，然后consumers來(lái)進(jìn)行消費(fèi)，如下圖：服務(wù)端(brokers)和客戶(hù)端(producer、consumer)TCP協(xié)議來(lái)完成。我們?yōu)镵afka提供了一個(gè)Java客戶(hù)端，但是也可以使用其他語(yǔ)言編寫(xiě)的客戶(hù)端。Topic讓我們首先深入理解Kafka提出一個(gè)次的抽象概念-Topic可以理解Topic是一個(gè)類(lèi)別的名稱(chēng)，所有的message發(fā)送到Topic下面。對(duì)于每一個(gè)Topic，kafka集群按照如下方式一個(gè)分區(qū)(Partition,可以將消息就理解為一個(gè)隊(duì)列Queue)日志文件:partition是一個(gè)有序的message序列，這些message按順序添加到一個(gè)叫做commitlog的文件中。每個(gè)partition中的消息都有一個(gè)唯一的編號(hào)，稱(chēng)之為offset，用來(lái)唯一標(biāo)示某個(gè)分區(qū)中的message。提示：每個(gè)partition，都對(duì)應(yīng)一個(gè)commit-log。一個(gè)partition中的messageoffset都是唯一的，但是不partition中的message的offset可能是相同的。kafka集群，在配置的時(shí)間范圍內(nèi)，所有的由producer生成的消息，而不管這些消息有沒(méi)有被消費(fèi)。例如日志保留logretention2天。kafka會(huì)維22天前的消息會(huì)被丟棄。kafka的性能與保留的數(shù)每個(gè)consumer是基于自己在commitlog中的消費(fèi)進(jìn)度(offset)來(lái)進(jìn)行工作的。在kafka中，offset由consumer來(lái)：一般情況下我們按照順序逐條消費(fèi)commitlog中的消息，當(dāng)然我可以通過(guò)指定offset來(lái)重復(fù)消費(fèi)某些消息，或者跳過(guò)某些消這意味kafka中的consumer對(duì)集群的影響是非常小的，添加一個(gè)或者減少一個(gè)consumer，對(duì)于集群或者其他consumer來(lái)說(shuō)，都是沒(méi)有影響的，因此每個(gè)consumer各自的offset。對(duì)log進(jìn)行分區(qū)（partitioned），log文件大小超過(guò)系統(tǒng)文件系統(tǒng)的限制時(shí)，可以自動(dòng)拆分。每個(gè)partition對(duì)應(yīng)的log都受到所在機(jī)器的文件系統(tǒng)大小的限制，但是一個(gè)Topic中是可以有很多分區(qū)的，因此可以處理任意數(shù)量log的partitions分布在kafka集群中不同的broker上，每個(gè)broker可以請(qǐng)求備份其他broker上partition上的數(shù)據(jù)。kafka集群支持配置一個(gè)partition針對(duì)每個(gè)partitionbroker起到“l(fā)eader”的作用，0個(gè)多個(gè)其他的broker作為“follwers”的作用。leader處理所有的針對(duì)這個(gè)partition的讀寫(xiě)請(qǐng)求，而followersleader的結(jié)果。如果這個(gè)leader失效了，其中的一個(gè)follower將會(huì)自動(dòng)的變成新的leader。每個(gè)broker都是自己所管理的partitionleader，同時(shí)又是其他broker所管理partitionsfollowers，kafka通過(guò)這種方式來(lái)達(dá)到負(fù)生產(chǎn)者將消息發(fā)送到topic中去，同時(shí)負(fù)責(zé)選擇將message發(fā)送到topic的哪一個(gè)partition中。通過(guò)round-robin做簡(jiǎn)單的負(fù)載均衡。也可以根據(jù)消息中的某一個(gè)2種：隊(duì)列queuing)和（publish-subscribe）在queuing模式中，多個(gè)consumer從服務(wù)器中數(shù)據(jù)，消息只會(huì)到達(dá)一個(gè)consumerpublish-subscribe模型中，消息會(huì)被廣播給所有的consumer。Kafka2種模式提供了一種consumer的抽象概念：consumergroup。每個(gè)consumer都要標(biāo)記自己屬于哪一個(gè)consumergroup。發(fā)布到topic中的message中message會(huì)被傳遞到consumergroup中的一個(gè)consumer實(shí)例。consumer實(shí)例可以運(yùn)行在不同的進(jìn)程上，也可以在不同的物理機(jī)器上。如果所有的consumer都位于同一個(gè)consumergroup下，這就類(lèi)似于傳統(tǒng)的模式，并在眾多的consumerinstance如果所有的consumer都有著自己唯一的consumergroup，這就類(lèi)似于傳統(tǒng)的publish-subscribe模型。更一般的情況是，通常一個(gè)topic會(huì)有幾個(gè)consumergroup，每個(gè)consumer都是一個(gè)邏輯上的訂閱者（logicalsubscriber）。每個(gè)consumergroup個(gè)consumerinstance組成，從而達(dá)到可擴(kuò)展和容災(zāi)的功能。這并沒(méi)有什么特殊的地方，僅僅是將publish-subscribe模型中的運(yùn)行在單個(gè)進(jìn)程上的consumers中的consumer替換成一個(gè)consumergroup。如下圖所示：2broker組成的kafka4個(gè)Parition(P0-P3)。這個(gè)集群由2個(gè)ConsumerGroupA2consumerinstances，而B(niǎo)有四個(gè).消費(fèi)順Kafka比傳統(tǒng)的消息系統(tǒng)有著更強(qiáng)的順序保證。在傳統(tǒng)的情況下，服務(wù)器按照順序保留消息到隊(duì)列，如果有多個(gè)consumer一個(gè)consumer是異步的，這可能會(huì)導(dǎo)致先消費(fèi)的consumer獲取到消息時(shí)間可能比后消費(fèi)的consumer獲取到消息的時(shí)間長(zhǎng)，導(dǎo)致不能保證順序性。這表明，當(dāng)進(jìn)consumer之間可能會(huì)失去順序性。消息系統(tǒng)通常會(huì)采取一種“exclusiveconsumer”的概念，來(lái)確保同一時(shí)間內(nèi)只有一個(gè)consumer能夠從隊(duì)列中進(jìn)行消費(fèi)，但是這實(shí)際上意味著在消息處理的過(guò)程中是不Kafka在這方面做的更好。通過(guò)Topic中并行度的概念，即partition，Kafka可以同時(shí)提供順序性保證和多個(gè)consumer同時(shí)消費(fèi)時(shí)的負(fù)載均衡。實(shí)現(xiàn)的原理是通過(guò)將一個(gè)topic中的partition分配給一個(gè)consumergroup中的不同consumerinstance。通過(guò)這種方式，我們可以保證一個(gè)partition在同一個(gè)時(shí)刻只有一個(gè)consumerinstance在消息，從而保證順序。雖然一個(gè)topic中有多個(gè)partition，但是一個(gè)consumergroupconsumerinstance，通過(guò)合理的分配依然能夠保證負(fù)載均衡。需要注意的是，一個(gè)consumergroup中的consumerinstance的數(shù)量不能比一個(gè)Topic中的partition的數(shù)量多。Kafka只在partition的范圍內(nèi)保證消息消費(fèi)的局部順序性，不能在同一個(gè)topic中的多個(gè)partition中保證總的消費(fèi)順序性。通常來(lái)說(shuō)，這已經(jīng)可以滿(mǎn)足大部分應(yīng)用過(guò)將topic的partition1，將consumergroup中的consumerinstance數(shù)1.從較高的層面上來(lái)說(shuō)的話(huà)，Kafka發(fā)送到一個(gè)Topic中的message會(huì)按照發(fā)送的順序添加到commitlog中。意思是，M1，M2由同一個(gè)producer發(fā)送，M1比M2發(fā)送的早的話(huà)，那么在commitlog中，M1的offset就會(huì)commit2offset小。一個(gè)consumer在commitlog中可以按照發(fā)送順序來(lái)消費(fèi)如果一個(gè)topic的備份因子replicationfactor)設(shè)置為N，那么KafkaN-1一個(gè)服務(wù)器，而在commitlog中的消息不會(huì)丟失安裝前的環(huán)境準(zhǔn)由于KafkaScala語(yǔ)言開(kāi)發(fā)的，運(yùn)行在JVM上，因此在安裝Kafka之前需要先安裝JDK。#yuminstalljava-1.8.0-openjdk*-kafka依賴(lài)zookeeper，所以需要先安裝#wgethttp://m #tar-zxvfzookeeper-3.4.12.tar.gz#cdzookeeper-#cpconf/zoo_sample.cfg啟動(dòng)#bin/zkServer.shstartconf/zoo.cfg&#bin/zkCli.sh#ls/#查看zk的根 1.1.0release版本，并解壓：#wget/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz#tar-xzfkafka_2.11-1.1.0.tgz#cdkafka_2.11-第二步：?jiǎn)?dòng)服現(xiàn)在來(lái)啟動(dòng)kafka啟動(dòng)語(yǔ)法：kafka-server-start.sh[-daemon]進(jìn)程運(yùn)行，否則ssh客戶(hù)端退出后，就會(huì)停止服務(wù)。(注意，在啟動(dòng)kafka使用linux主機(jī)名關(guān)聯(lián)的ip地址，所以需要把主機(jī)名和linux的iphost里，用vim#bin/kafka-server-start.sh-daemon我們進(jìn)入 通過(guò)zookeeper客戶(hù)端查看下zookeeper ##ls/#查看zk的 kafka相關(guān)節(jié)ls/brokers/ids查看kafka第三步：創(chuàng)現(xiàn)在我們來(lái)創(chuàng)建一個(gè)名字為“test”Topic，這個(gè)topic只有一個(gè)partition，并且備1：#bin/kafka-topics.sh--create--zookeeperlocalhost:2181--replication-factor1--partitions1--topictest現(xiàn)在我們可以通過(guò)以下命令來(lái)查看kafka中目前存在的topic#bin/kafka-topics.sh--list--zookeeperlocalhost:2181除了我們通過(guò)手工的方式創(chuàng)建Topic，我們可以配置broker，當(dāng)producer發(fā)布一個(gè)消息某個(gè)指定的Topic，但是這個(gè)Topic并不存在時(shí)，就自動(dòng)創(chuàng)建。第四步：發(fā)送消kafka自帶了一個(gè)producer命令客戶(hù)端，可以從本地文件中內(nèi)容，或者我們也首先我們要運(yùn)行發(fā)布消息的，然后在命令中輸入要發(fā)送的消息的內(nèi)容#bin/kafka-console-producer.sh--broker-listlocalhost:9092--topic>thisisa>thisisaanother第五步：消費(fèi)消對(duì)于consumer，kafka同樣也攜帶了一個(gè)命令行客戶(hù)端，會(huì)將獲取到內(nèi)容在命令中#bin/kafka-console-consumer.sh--zookeeperlocalhost:2181--topictest #bin/kafka-console-consumer.sh--bootstrap-serverlocalhost:9092--consumer-propertygroup.id=testGroup--consumer-propertyclient.id=consumer-1--topic 入的內(nèi)容，很快就會(huì)在consumer的終端窗口上顯示出來(lái)。#bin/kafka-consumer-groups.sh--bootstrap-serverlocalhost:9092--list--new-#bin/kafka-consumer-groups.sh--bootstrap-serverlocalhost:9092--describe--group#bin/kafka-console-consumer.sh--bootstrap-serverlocalhost:9092--whiist單播消一條消息只能被某一個(gè)消費(fèi)者消費(fèi)的模式，類(lèi)似queue模式，只需讓所有消費(fèi)者#bin/kafka-console-consumer.sh--bootstrap-serverlocalhost:9092--consumer-propertygroup.id=testGroup--topictest多播消一條消息能被多個(gè)消費(fèi)者消費(fèi)的模式，類(lèi)似publish-subscribe模式費(fèi)，針對(duì)Kafka同testGroup-2消費(fèi)#bin/kafka-console-consumer.sh--bootstrap-serverlocalhost:9092--consumer-propertygroup.id=testGroup-2--topictest第六步：kafka集群配到目前為止，我們都是在一個(gè)單節(jié)點(diǎn)上運(yùn)行broker，這并沒(méi)有什么意思。對(duì)于kafka來(lái)說(shuō)，一個(gè)單獨(dú)的broker意味著kafka集群中只有一個(gè)接點(diǎn)。要想增加kafka集群中的節(jié)點(diǎn)數(shù)量，只需要多啟動(dòng)幾個(gè)broker實(shí)例即可。為了有更好的理解，現(xiàn)在我們?cè)谝慌_(tái)機(jī)器上同時(shí)啟動(dòng)三個(gè)broker實(shí)例。2broker的配置文件：#cpconfig/pertiesconfig/perties#cpconfig/pertiesconfig/pertiesbrkrd屬性在kaka集群中必須要是唯一的。我們需要重新指定t和lg目錄，因?yàn)槲覀兪窃谕慌_(tái)機(jī)器上運(yùn)行多個(gè)實(shí)例。如果不進(jìn)行修改的話(huà)，smer只能獲取到一個(gè)istae實(shí)例的信息，或者是相互之間的數(shù)據(jù)會(huì)被影響。目前我們已經(jīng)有一個(gè)zookeeperbroker實(shí)例在運(yùn)行了，現(xiàn)在我們只需要2broker實(shí)例即可：bin/kafka-server-start.shconfig/perties&bin/kafka-server-start.shconfig/perties&現(xiàn)在我們創(chuàng)建一個(gè)新的topic3：#bin/kafka-topics.sh--create--zookeeperlocalhost:2181--replication-factor3--partitions1--topicmy-replicated-topic3個(gè)備份因子的topic，但是到底是哪一個(gè)broker在為這個(gè)topic提供服務(wù)呢(因?yàn)槲覀冎挥幸粋€(gè)分區(qū)，所以肯定同時(shí)只有一個(gè)broker在處理這個(gè)topic)？#bin/kafka-topics.sh--describe--zookeeperlocalhost:2181--topicmy-replicated-partition的信息。因?yàn)槟壳拔覀冎挥幸粋€(gè)partitionpartition的信息只有l(wèi)eader節(jié)點(diǎn)負(fù)責(zé)給定partition的所有讀寫(xiě)請(qǐng)求。leader信息可以在zookeeper里 查看replicas表示某個(gè)partition在哪幾個(gè)broker上存在備份。不管這個(gè)幾點(diǎn)是不是”leader“，甚isr是replicaspartition現(xiàn)在我們的案例中，0號(hào)節(jié)點(diǎn)是leader，即使用perties啟動(dòng)的那個(gè)進(jìn)我們可以運(yùn)行相同令查看之前創(chuàng)建的名稱(chēng)為”test“的#bin/kafka-topics.sh--describe--zookeeperlocalhost:2181--topic沒(méi)有