kafka設計原理分析集群搭建與使用

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