Kafka多節(jié)點(diǎn)分布式集群搭建實(shí)現(xiàn)過程詳解
上一篇分享了單節(jié)點(diǎn)偽分布式集群搭建方法，本篇來分享一下多節(jié)點(diǎn)分布式集群搭建方法。多節(jié)點(diǎn)分布式集群結(jié)構(gòu)如下圖所示：
為了方便查閱，本篇將和上一篇一樣從零開始一步一步進(jìn)行集群搭建。
一、安裝Jdk
具體安裝步驟可參考linux安裝jdk。
二、安裝與配置zookeeper
下載地址：https://www-us.apache.org/dist/zookeeper/stable/
下載二進(jìn)制壓縮包zookeeper-3.4.14.tar.gz，然后上傳到linux服務(wù)器指定目錄下，本次上傳目錄為/software，然后執(zhí)行如下命令安裝：
cd /software
tar -zxvf zookeeper-3.4.14.tar.gz
mv zookeeper-3.4.14 /usr/local/zookeeper
cd /usr/local/zookeeper/conf
mv zoo_sample.cfg zoo1.cfg
編輯zoo1.cfg,配置相關(guān)參數(shù)如下：
tickTime=2000
initLimit=5
syncLimit=2
dataDir=/usr/local/zookeeper/data/zookeeper1
clientPort=2181
server.1=192.168.184.128:2888:3888
server.2=192.168.184.128:2889:3889
server.3=192.168.184.128:2890:3890
其中：
tickTime：Zookeeper最小的時(shí)間單位，用于丈量心跳和超時(shí)時(shí)間，一般設(shè)置默認(rèn)值2秒；
initLimit：指定follower節(jié)點(diǎn)初始時(shí)連接leader節(jié)點(diǎn)的最大tick此處，設(shè)置為5，表示follower必須在5xtickTime即10秒內(nèi)連接上leader,否則視為超時(shí)；
syncLimit：設(shè)定follower節(jié)點(diǎn)與leader節(jié)點(diǎn)進(jìn)行同步的最大時(shí)間，設(shè)置為2，表示最大時(shí)間為2xtickTime即4秒時(shí)間；
dataDir：Zookeeper會(huì)在內(nèi)存中保存系統(tǒng)快照，并定期寫入該路徑指定的文件夾中，生產(chǎn)環(huán)境需要特別注意該文件夾的磁盤占用情況；
clientPort：Zookeeper監(jiān)聽客戶端連接的端口號(hào)，默認(rèn)為2181，同一服務(wù)器上不同實(shí)例之間應(yīng)該有所區(qū)別；
server.X=host:port1:port2:此處X的取值范圍在1~255之間，必須是全局唯一的且和myid文件中的數(shù)字對(duì)應(yīng)（myid文件后面說明）,host是各個(gè)節(jié)點(diǎn)的主機(jī)名，port1通常是2888，用于使follower節(jié)點(diǎn)連接leader節(jié)點(diǎn)，port2通常是3888，用于leader選舉，zookeeper在不同服務(wù)器上的時(shí)候，不同zookeeper服務(wù)器的端口號(hào)可以重復(fù)，在同一臺(tái)服務(wù)器上的時(shí)候需要有所區(qū)別。
1.配置zoo.cfg文件
單節(jié)點(diǎn)安裝zookeeper的時(shí)候，僅有一份zoo.cfg文件，多節(jié)點(diǎn)安裝的時(shí)候，每個(gè)zookeeper服務(wù)器就應(yīng)該有一個(gè)zoo.cfg配置文件。如果在一臺(tái)服務(wù)器安裝zookeeper多實(shí)例集群，則需要在conf目錄下分別配置每個(gè)實(shí)例的zoo.cfg，同時(shí)創(chuàng)建每個(gè)zookeeper實(shí)例自己的數(shù)據(jù)存儲(chǔ)目錄。本次在一臺(tái)服務(wù)器上配置多個(gè)zookeeper實(shí)例，執(zhí)行如下命令創(chuàng)建數(shù)據(jù)存儲(chǔ)目錄并復(fù)制配置文件：
mkdir -p /usr/local/zookeeper/data/zookeeper1
mkdir -p /usr/local/zookeeper/data/zookeeper2
mkdir -p /usr/local/zookeeper/data/zookeeper3
cd /usr/local/zookeeper/conf/
cp zoo1.cfg zoo2.cfg
cp zoo1.cfg zoo3.cfg
復(fù)制后分別修改zoo2.cfg,zoo3.cfg中的配置，修改后的配置如下：
zoo1.cfg的配置如下:
zoo2.cfg的配置如下:
zoo3.cfg中的配置如下：
2.myid文件創(chuàng)建與配置
前面提到zoo.cfg文件中的server.X中的X應(yīng)該與myid中的數(shù)字相對(duì)應(yīng)。除此之外，myid文件必須存放在每個(gè)zookeeper實(shí)例的data目錄下，對(duì)應(yīng)本次安裝應(yīng)該位于/usr/local/zookeeper/data/zookeeper1，2,3目錄下，執(zhí)行如下命令進(jìn)行配置：
echo '1' > /usr/local/zookeeper/data/zookeeper1/myid
echo '2' > /usr/local/zookeeper/data/zookeeper2/myid
echo '3' > /usr/local/zookeeper/data/zookeeper3/myid
3.啟動(dòng)zookeeper服務(wù)器
使用如下命令啟動(dòng)zookeeper集群:
cd /usr/local/zookeeper/bin/
./zkServer.sh start ../conf/zoo1.cfg
./zkServer.sh start ../conf/zoo2.cfg
./zkServer.sh start ../conf/zoo3.cfg
啟動(dòng)后，使用如下命令查看集群狀態(tài):
cd /usr/local/zookeeper/bin/
./zkServer.sh status ../conf/zoo1.cfg./zkServer.sh status ../conf/zoo2.cfg./zkServer.sh status ../conf/zoo3.cfg
回顯信息如下：
可以看到有兩個(gè)follower節(jié)點(diǎn)，一個(gè)leader節(jié)點(diǎn)。
三、安裝與配置kafka集群
下載地址：http://kafka.apache.org/downloads.html
1.數(shù)據(jù)目錄和配置文件創(chuàng)建
目前最新版本是2.2.0,本次下載2.1.1版本的安裝包，然后上傳壓縮包到服務(wù)器指定目錄，本次上傳目錄為/software，然后執(zhí)行以下命令進(jìn)行安裝:
tar -zxvf kafka_2.12-2.1.1.tgz
mv kafka_2.12-2.1.1 /usr/local/kafka
mkdir -p /usr/local/kafka/logs/kafka1
mkdir -p /usr/local/kafka/logs/kafka2
mkdir -p /usr/local/kafka/logs/kafka3
cd /usr/local/kafka/config/
mv server.properties server1.properties
通過執(zhí)行上面的命令，我們在/usr/local/kafka/logs文件夾中創(chuàng)建了kafka1,kafka2,kafka3三個(gè)文件夾用于存放三個(gè)kafka實(shí)例的數(shù)據(jù)，同時(shí)將/usr/local/kafka/config/文件夾下的server.properties重命名為server1.properties用于配置kafka的第一個(gè)實(shí)例。
2.配置屬性文件
接下來配置server1.properties文件，主要配置參數(shù)如下：
broker.id=1：設(shè)置kafka broker的id，本次分別為1,2,3；
delete.topic.enable=true:開啟刪除topic的開關(guān)；
listeners=PLAINTEXT://192.168.184.128:9092：設(shè)置kafka的監(jiān)聽地址和端口號(hào)，本次分別設(shè)置為9092,9093,9094；
log.dirs=/usr/local/kafka/logs/kafka1：設(shè)置kafka日志數(shù)據(jù)存儲(chǔ)路徑；
zookeeper.connect=192.168.184.128:2181,192.168.184.128:2182,192.168.184.128:2183：設(shè)置kafka連接的zookeeper訪問地址,集群環(huán)境需要配置所有zookeeper的訪問地址;
unclean.leader.election.enable=false：為true則代表允許選用非isr列表的副本作為leader，那么此時(shí)就意味著數(shù)據(jù)可能丟失，為false的話，則表示不允許，直接拋出NoReplicaOnlineException異常，造成leader副本選舉失敗。
zookeeper.connection.timeout.ms=6000：設(shè)置連接zookeeper服務(wù)器超時(shí)時(shí)間為6秒。
配置完成后，復(fù)制server1.properties兩份分別用于配置kafka的第二個(gè)，第三個(gè)節(jié)點(diǎn)：
修改修改其中的broker.id 以及l(fā)isteners、log.dirs的配置為第二個(gè)，第三個(gè)節(jié)點(diǎn)的配置，最終各個(gè)配置文件配置如下:
server1.properties配置：
server2.properties配置：
server3.properties配置:
3.啟動(dòng)kafka
通過如下命令啟動(dòng)kafka集群:
cd /usr/local/kafka/bin/
./kafka-server-start.sh -daemon ../config/server1.properties
./kafka-server-start.sh -daemon ../config/server2.properties
./kafka-server-start.sh -daemon ../config/server3.properties
使用java的命令jps來查看kafka進(jìn)程:jps |grep -i kafka
說明kafak啟動(dòng)正常，至此kafka集群搭建完成。本次使用一臺(tái)服務(wù)器作為演示，如果需要在多個(gè)服務(wù)器上配置集群，配置方法和以上類似，只是不需要像上面那樣配置多個(gè)數(shù)據(jù)目錄和配置文件，每臺(tái)服務(wù)器的配置保持相同，并且注意在防火墻配置端口號(hào)即可。
最后，如果需要遠(yuǎn)程訪問kafka集群，則需要在防火墻中開通9092、9093、9094端口的訪問權(quán)限。
四、測試
1.topic創(chuàng)建與刪除
首先創(chuàng)建一個(gè)測試topic，名為testTopic，為了充分利用3個(gè)實(shí)例（服務(wù)器節(jié)點(diǎn)），創(chuàng)建3個(gè)分區(qū)，每個(gè)分區(qū)都分配3個(gè)副本，命令如下：
cd /usr/local/kafka/bin/
./kafka-topics.sh --zookeeper 192.168.184.128:2181 192.168.184.128:2182 192.168.184.128:2183 --create --topic testTopic --partitions 3 --replication-factor 3
回顯Created topic "testTopic".則表明testTopic創(chuàng)建成功。執(zhí)行如下命令進(jìn)行驗(yàn)證并查看testTopic的信息:
./kafka-topics.sh --zookeeper 192.168.184.128:2181 192.168.184.128:2182 192.168.184.128:2183 --list testTopic
./kafka-topics.sh --zookeeper 192.168.184.128:2181 192.168.184.128:2182 192.168.184.128:2183 --describe --topic testTopic
以上幾條命令回顯信息如下：
接下來測試topic刪除，使用如下命令進(jìn)行刪除:
./kafka-topics.sh --zookeeper 192.168.184.128:2181 192.168.184.128:2182 192.168.184.128:2183 --delete --topic testTopic
執(zhí)行該條命令后，回顯信息如下：
可以看到，testTopic已經(jīng)被標(biāo)記為刪除，同時(shí)第二行提示表明當(dāng)配置了delete.topic.enable屬性為true的時(shí)候topic才會(huì)刪除，否則將不會(huì)被刪除，本次安裝的時(shí)候該屬性設(shè)置的值為true。
2.測試消息發(fā)送與消費(fèi)
首先使用第一步topic創(chuàng)建命令，先創(chuàng)建testTopic這個(gè)topic，然后進(jìn)行消息發(fā)送與消費(fèi)測試。
控制臺(tái)測試消息發(fā)送與消費(fèi)需要使用kafka的安裝目錄/usr/local/kafka/bin下的kafka-console-producer.sh來發(fā)送消息，使用kafka-console-consumer.sh來消費(fèi)消息。因此本次打開兩個(gè)控制臺(tái)，一個(gè)用于執(zhí)行kafka-console-producer.sh來發(fā)送消息，另一個(gè)用于執(zhí)行kafka-console-consumer.sh來消費(fèi)消息。
消息發(fā)送端命令:
cd /usr/local/kafka/bin
./kafka-console-producer.sh --broker-list 192.168.184.128:9092,192.168.184.128:9093,192.168.184.128:9094 --topic testTopic
消息接收端命令:
cd /usr/local/kafka/bin
./kafka-console-consumer.sh --bootstrap-server 192.168.184.128:9092,192.168.184.128:9093,192.168.184.128:9094 --topic testTopic --from-beginning
當(dāng)發(fā)送端和接收端都登錄后，在發(fā)送端輸入需要發(fā)送的消息并回車，在接收端可以看到剛才發(fā)送的消息：
發(fā)送端：
接收端：
以上就是簡單地生產(chǎn)消息與消費(fèi)消息的測試，在測試消費(fèi)消息的時(shí)候時(shí)候，命令里邊加了個(gè)參數(shù)--from-beginning表示接收該topic從創(chuàng)建開始的所有消息。
3.生產(chǎn)者吞吐量測試
對(duì)于任何一個(gè)消息引擎而言，吞吐量是一個(gè)至關(guān)重要的性能指標(biāo)。對(duì)于Kafka而言，它的吞吐量指每秒能夠處理的消息數(shù)或者字節(jié)數(shù)。kafka為了提高吞吐量，采用追加寫入方式將消息寫入操作系統(tǒng)的頁緩存，讀取的時(shí)候從頁緩存讀取，因此它不直接參與物理I/O操作，同時(shí)使用以sendfile為代表的零拷貝技術(shù)進(jìn)行數(shù)據(jù)傳輸提高效率。
kafka提供了kafka-producer-perf-test.sh腳本用于測試生產(chǎn)者吞吐量，使用如下命令啟動(dòng)測試：
cd /usr/local/kafka/bin
./kafka-producer-perf-test.sh --topic testTopic --num-records 50000 --record-size 200 --throughput -1 --producer-props bootstrap.servers=192.168.184.128:9092,192.168.184.128:9093,192.168.184.128:9094 acks=-1
以上回顯信息表明這臺(tái)服務(wù)器上每個(gè)producer每秒能發(fā)送6518個(gè)消息，平均吞吐量是1.24MB/s，平均延遲2.035秒，最大延遲3.205秒，平均有50%的消息發(fā)送需要2.257秒，95%的消息發(fā)送需要3.076秒，99%的消息發(fā)送需要3.171秒，99.9%的消息發(fā)送需要3.205秒。
4.消費(fèi)者吞吐量測試
與生產(chǎn)者吞吐量測試類似，kafka提供了kafka-consumer-perf-test.sh腳本用于消費(fèi)者吞吐量測試，可以執(zhí)行以下命令進(jìn)行測試：
cd /usr/local/kafka/bin
./kafka-consumer-perf-test.sh --broker-list 192.168.184.128:9092,192.168.184.128:9093,192.168.184.128:9094 --messages 50000 --topic testTopic
以上是測試50萬條消息的consumer吞吐量，結(jié)果表明該consumer在1秒總共消費(fèi)了9.5366MB消息。
以上就是kafka集群的搭建以及測試，如有錯(cuò)誤之處，煩請(qǐng)指正。
參考資料：《Apache kafka實(shí)戰(zhàn)》
以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
idea中maven項(xiàng)目模塊變成灰色原因及解決方案
這篇文章主要介紹了idea中maven項(xiàng)目模塊變成灰色原因及解決方案,文中通過圖文結(jié)合的方式給大家講解的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作有一定的幫助,需要的朋友可以參考下2024-03-03
解讀jdk動(dòng)態(tài)代理為什么必須實(shí)現(xiàn)接口
這篇文章主要介紹了解讀jdk動(dòng)態(tài)代理為什么必須實(shí)現(xiàn)接口問題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2023-02-02
這篇文章主要介紹了java中對(duì)HashMap的put過程解讀，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2023-03-03
SpringBoot整合Druid數(shù)據(jù)源過程詳解
這篇文章主要介紹了SpringBoot整合Druid數(shù)據(jù)源過程詳解,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-12-12
深入理解Java中的并發(fā)工具類CountDownLatch
CountDownLatch?作為?Java?中的一個(gè)同步工具類，用于在多線程間實(shí)現(xiàn)協(xié)調(diào)和控制，本文主要來和大家講解一下JUC?工具類?CountDownLatch的使用，需要的可以參考一下2023-07-07
springboot整合mybatis的超詳細(xì)過程(配置模式+注解模式)
這篇文章主要介紹了springboot整合mybatis的詳細(xì)過程(配置模式+注解模式),這里我使用的是配置模式+注解模式所以需要配置全局文件，本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，需要的朋友可以參考下2022-04-04
基于Java實(shí)現(xiàn)Json文件轉(zhuǎn)換為Excel文件
這篇文章主要為大家詳細(xì)介紹了如何利用Java實(shí)現(xiàn)Json文件轉(zhuǎn)換為Excel文件，文中的示例代碼講解詳細(xì)，具有一定的借鑒價(jià)值，需要的可以參考一下2022-12-12
Java MongoDB數(shù)據(jù)庫連接方法梳理
MongoDB作為一種介于關(guān)系型數(shù)據(jù)庫和非關(guān)系型數(shù)據(jù)庫之間的產(chǎn)品，它可以提供可擴(kuò)展的高性能的數(shù)據(jù)存儲(chǔ)解決方案，近些年來受到了開發(fā)者的喜愛2022-08-08
JAVA Iterator接口與增強(qiáng)for循環(huán)的實(shí)現(xiàn)
這篇文章主要介紹了JAVA Iterator接口與增強(qiáng)for循環(huán)的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-11-11
Java經(jīng)典設(shè)計(jì)模式之策略模式原理與用法詳解
這篇文章主要介紹了Java經(jīng)典設(shè)計(jì)模式之策略模式,簡單說明了策略模式的概念、原理并結(jié)合實(shí)例形式分析了java策略模式的具有用法與相關(guān)注意事項(xiàng),需要的朋友可以參考下2017-08-08