07-kafka集群搭建與使用諸葛

1、 對于熟悉JMS（JavaMessageService）規(guī)范的同學(xué)來說，消息系統(tǒng)已經(jīng)不是什么新 概念了(例如ActiveMQ，RabbitMQ 等)。Kafka 對于熟悉JMS（JavaMessageService）規(guī)范的同學(xué)來說，消息系統(tǒng)已經(jīng)不是什么新 概念了(例如ActiveMQ，RabbitMQ 等)。Kafka來說，Kafka 借鑒了JMS 規(guī)范，但是確并沒有完全遵循JMS 規(guī)范kafka 是一個分布式的，分區(qū)的消息(稱之為commitlog)服務(wù)。它提供一個消 Topic:Kafka 按照Topic分類消 發(fā)布(publish)消息到Topic的進(jìn)程稱之為生產(chǎn)者 訂閱(subscri

2、be)Topic并且處理Topic BrokerKafka因此，從一個較高的層面上來看，producers通過網(wǎng)絡(luò)發(fā)送消息到Kafka集群，然后consumers 來進(jìn)行消費，如下圖：們?yōu)镵afka 提供了一個Java 客戶端，但是也可以使用其他語言編寫的客戶端。Topic讓首先深入理解Kafka 可以理解Topic 是一個類別的名稱，所有的們?yōu)镵afka 提供了一個Java 客戶端，但是也可以使用其他語言編寫的客戶端。Topic讓首先深入理解Kafka 可以理解Topic 是一個類別的名稱，所有的message 發(fā)送到Topic 個Topic，kafka集群按照如下方式 隊列Queue)日志

3、文件:partition 是一個有序的message 序列，這些message 按順序添加到一個叫做 commit log 的文件中。每個partition 中的消息都有一個唯一的 ，稱之為 提示：每個partition，都對應(yīng)一個commit-log。一個partition 中的 messageoffset都是唯一的，但是partition中的message的offset 可能是相同的kafka 所有的由producer 每個consumer 是基于自己在commit log 中的消費進(jìn)度(offset)kafka中，offset由consumer按照順序逐條消費log 中的消息，當(dāng)然我可以

4、通過指定offset 來重復(fù)消費某些消息，或者跳過某些消些消息有沒有被消費。例如日志保留logretention2天。kafka會維2 2 天前的消息會被丟棄。kafka 這意味kafka中的consumer對集群的影響是非常小的，添加一個或者減少一個 consumer，對于集群或者其他consumer 來說，都是沒有影響的，因此每個consumer offset對log 進(jìn)行分區(qū)（partitioned這意味kafka中的consumer對集群的影響是非常小的，添加一個或者減少一個 consumer，對于集群或者其他consumer 來說，都是沒有影響的，因此每個consumer offse

5、t對log 進(jìn)行分區(qū)（partitioned），有以下目的。首先，當(dāng)log 文件大小超過系統(tǒng)文 件系統(tǒng)的限制時，可以自動拆分。每個partition 對應(yīng)的log 都受到所在機器的文件系統(tǒng)大小的限制，但是一個Topic中是可以有很多分區(qū)的，因此可以處理任意數(shù)量log 的partitions 分布在kafka 集群中不同的broker 上，每個broker 可以請求備份 其他broker 上partition 上的數(shù)據(jù)。kafka 集群支持配置一個partition 針對每個partition，都有一個broker 起到“l(fā)eader”的作用，0 個多個其他的 作為“follwers”的作用。l

6、eader 處理所有的針對這個partition 會自動的變成新的leader。broker都是自己所管理的partition leader，同 時又是其他 broker 所管理 partitions 的 followers，kafka 通過這種方式來達(dá)到負(fù)載生產(chǎn)者將消息發(fā)送到topic 中去，同時負(fù)責(zé)選擇將message 發(fā)送到topic 的哪一個 partition 中。通過round-robin在queuing 模式中，多個consumer 從服務(wù)器中consumer。在publish-subscribe模型中，消息會被廣播給所有的consumer。Kafka 基于這 2 種模式提供了一

7、種consumer 的抽象概念：consumer group。每個consumer 都要標(biāo)記自己屬于哪一個consumergroup。發(fā)布到topic 中的 message 中message 會被傳遞到consumer group 中的一個consumer 實例。 consumer 實例可以運行在不同的進(jìn)程上，也可以在不同的物理機器上。如果所有的consumer 都位于同一個consumer group 下，這就類似于傳統(tǒng)的 模式，并在眾多的consumer instance 如果所有的consumer 都有著自己唯一的consumer group，這就類似于傳統(tǒng)的 publish-subsc

8、ribe 模型。更一般的情況是，通常一個topic 會有幾個consumer group，每個consumer 都是一個邏輯上的訂閱者（logicalsubscriber）。每個consumergroup leader 的結(jié)果。如果這個leader 失效了，其中的一個follower個consumerinstance地方，僅僅是將publish-subscribe 模型中的運行在單個進(jìn)程上的consumers 中的 consumer 替換成一個consumer group。如下圖所示：個consumerinstance地方，僅僅是將publish-subscribe 模型中的運行在單個進(jìn)程上的

9、consumers 中的 consumer 替換成一個consumer group。如下圖所示：2broker組成的kafka4個Parition(P0-P3)2個ConsumerGroup，A 有2個consumer，而B 有四個消費順Kafka 比傳統(tǒng)的消息系統(tǒng)有著更強的順序保證。在傳統(tǒng)的情況下，服務(wù)器按照順序 保留消息到隊列，如果有多個consumer 來消費隊列中的消息，服務(wù)器 會接受消息一個consumer 是異步的，這可能會導(dǎo)致先消費的consumer 獲取到消息時間可能比后消費的consumer 獲取到消息的時間長，導(dǎo)致不能保證順序性。這表明，當(dāng)進(jìn)行并行的消費的時候，消息在多個

10、consumer 之間可能會失去順序性。消息系統(tǒng)通常會采取一種“ exclusive consumer”的概念，來確保同一時間內(nèi)只有一個 consumer 能夠從隊列中進(jìn)行消費，但是這實際上意味著在消息處理的過程中是不Kafka 在這方面做的更好。通過Topic 中并行度的概念，即partition，Kafka 可以同 時提供順序性保證和多個consumer 同時消費時的負(fù)載均衡。實現(xiàn)的原理是通過將一個topic中的partition 分配給一個consumer group 中的不同consumer instance可以保證一個partition 在同一個時刻只有一個consumer inst

11、ance 在消息，從而保證順序。雖然一個topic 中有多個partition，但是一個 負(fù)載均衡。需要注意的是，一個consumer group 中的consumer instance 的數(shù)量不能比一個Topic 中的partition 的數(shù)量多。Kafka 只在partition 的范圍內(nèi)保證消息消費的局部順序性，不能在同一個topic 中的多個partition 過將topic的partition1，將consumergroup中的consumerinstanceKafka 只在partition 的范圍內(nèi)保證消息消費的局部順序性，不能在同一個topic 中的多個partition 過將

12、topic的partition1，將consumergroup中的consumerinstance數(shù) 量也設(shè)置為 1.從較高的層面上來說的話，Kafka發(fā)送到一個Topic 中的message 會按照發(fā)送的順序添加到commit log 中。意思是， 如果消息 M1，M2 由同一個producer 發(fā)送，M1 比M2 發(fā)送的早的話，那么在 commit log 中，M1 的offset 就會commit 2 offset 小。一個consumer 在commit log 中可以按照發(fā)送順序來消費如果一個topic的備份因子replicationfactor)設(shè)置為N，那么KafkaN-1 在c

13、ommitlog 安裝前的環(huán)境準(zhǔn)備由于KafkaScala 語言開發(fā)的，運行在JVM 上，因此在安裝Kafka 之前需要先安裝JDK。#yuminstalljava-1.8.0-openjdk*-kafka 依賴zookeeper，所以需要先安裝# wget HYPERLINK http:/m/apache/zookeeper/stable/zookeeper-3.4.12.tar.gz #tar-zxvfzookeeper-#cdzookeeper-#cp啟動le.cfg #bin/zkServer.shstartconf/zoo.cfg#ls #查看zk 第一步安裝1.1.0release

14、# /apache/kafka/1.1.0/kafka 2.11-#tar-xzfkafka_2.11-#cdkafka_2.11-第二步：啟動服務(wù)現(xiàn)在來啟動kafka 語法：kafka-server-start.shdaemon可以看到，1.1.0release# /apache/kafka/1.1.0/kafka 2.11-#tar-xzfkafka_2.11-#cdkafka_2.11-第二步：啟動服務(wù)現(xiàn)在來啟動kafka 語法：kafka-server-start.shdaemon可以看到，roperties 的配置路徑是一個強制的參數(shù)，-daemon 表示以進(jìn)程運行，否則ssh 客戶

15、端退出后，就會停止服務(wù)。(注意，在啟動kafka 時會 使用linux 主機名關(guān)聯(lián)的ip 地址，所以需要把主機名配置到本地host 里，用vim#bin/kafka-server-start.sh-daemon進(jìn)入zookeeper目錄通過zookeeper客戶端查看下zookeeper的目錄樹 # bin/zkCli.shls #查看zk 的根目錄kafka ls/brokers/ids查看kafka 第三步：創(chuàng)建來創(chuàng)建一個名字為“test”的Topic，這個topic只有一個partition，并且備 份因子也設(shè)置為 1：#bin/kafka-topics.sh-create-zookee

16、perlocalhost:2181-replication-factor1- partitions 1 -topic test現(xiàn)可以通過以下命令來查看kafka 中目前存在的#bin/kafka-topics.sh-list-zookeeper通過手工的方式創(chuàng)建可以配置broker，當(dāng)producer 消息某個指定的Topic，但是這個Topic 第四步：發(fā)送消息kafka 自帶了一個producer 也 thisis athisisaanother第五步：消費消息對于consumer，kafka同樣也攜帶了一個命令行客戶端，會將獲取到內(nèi)容在命令中#kafka 自帶了一個producer 也

17、thisis athisisaanother第五步：消費消息對于consumer，kafka同樣也攜帶了一個命令行客戶端，會將獲取到內(nèi)容在命令中#bin/kafka-console-consumer.sh-zookeeperlocalhost:2181-topictest-令，你將會看到在producer終端輸 入的內(nèi)容，很快就會在consumer 的終端窗口上顯示出來。令都有一些附加的不攜帶任何參數(shù)運行命令的候，將會顯示出這個命令的詳細(xì)用法第六步：kafka 集群配都是在一個單節(jié)點上運行brokerkafka 來說，一個單獨的broker 意味著kafka 集群中只有一個接點。要想增加 集群

18、中的節(jié)點數(shù)量，只需要多啟動幾個broker 在一臺機器上同時啟動三個broker 實例#cpropertiesconfig/server-#cp broker.id屬性在kafka集群中必須要是唯一的需要重新指定port 和log 只能獲取到一個instance broker.id屬性在kafka集群中必須要是唯一的需要重新指定port 和log 只能獲取到一個instance 已經(jīng)有一個zookeeperbroker2broker#bin/kafka-server-start.shconfig/perties#bin/kafka-server-start.shconfig/perties現(xiàn)

19、#bin/kafka-topics.sh-create-zookeeperlocalhost:2181-replication-factor3- partitions 1 -topic my-replicated-topic3 個備份因子的topic個broker 在為這個topic 提供服務(wù)呢(因為 一個broker 在處理這個topic)？#bin/kafka-topics.sh-describe-zookeeperlocalhost:2181-topicmy-replicated-partition 的信息。因為目前只有一個partition，因此關(guān)于partition 的信息只有 leader 節(jié)點負(fù)責(zé)給定partition 的所有讀寫請求。如果一個topic 有多個partitions，那么每 個節(jié)點都