上一篇分享了單節點偽分布式集群搭建方法，本篇來分享一下多節點分布式集群搭建方法。多節點分布式集群結構如下圖所示：
為了方便查閱，本篇將和上一篇一樣從零開始一步一步進行集群搭建。
一、安裝Jdk
具體安裝步驟可參考 linux安裝jdk。
二、安裝與配置zookeeper
下載地址：https://www-us.apache.org/dist/zookeeper/stable/
下載二進制壓縮包 zookeeper-3.4.14.tar.gz，然后上傳到linux服務器指定目錄下，本次上傳目錄為 /software ，然后執行如下命令安裝：
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 ,配置相關參數如下：
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最小的時間單位，用於丈量心跳和超時時間，一般設置默認值2秒；
initLimit ：指定follower節點初始時連接leader節點的最大tick此處，設置為5，表示follower必須在 5xtickTime 即10秒內連接上leader,否則視為超時；
syncLimit ：設定follower節點與leader節點進行同步的最大時間，設置為2，表示最大時間為 2xtickTime 即4秒時間；
dataDir ：Zookeeper會在內存中保存系統快照，並定期寫入該路徑指定的文件夾中，生產環境需要特別注意該文件夾的磁盤占用情況；
clientPort ：Zookeeper監聽客戶端連接的端口號，默認為2181，同一服務器上不同實例之間應該有所區別；
server.X=host:port1:port2 :此處X的取值范圍在1~255之間，必須是全局唯一的且和myid文件中的數字對應（myid文件后面說明）,host是各個節點的主機名，port1通常是2888，用於使follower節點連接leader節點，port2通常是3888，用於leader選舉，zookeeper在不同服務器上的時候，不同zookeeper服務器的端口號可以重復，在同一台服務器上的時候需要有所區別。
1.配置zoo.cfg文件
單節點安裝zookeeper的時候，僅有一份zoo.cfg文件，多節點安裝的時候，每個zookeeper服務器就應該有一個zoo.cfg配置文件。如果在一台服務器安裝zookeeper多實例集群，則需要在conf目錄下分別配置每個實例的zoo.cfg，同時創建每個zookeeper實例自己的數據存儲目錄。本次在一台服務器上配置多個zookeeper實例，執行如下命令創建數據存儲目錄並復制配置文件：
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
復制后分別修改 zoo2.cfg , zoo3.cfg 中的配置，修改后的配置如下：
zoo1.cfg的配置如下:
zoo2.cfg的配置如下:
zoo3.cfg中的配置如下：
2.myid文件創建與配置
前面提到zoo.cfg文件中的server.X中的X應該與myid中的數字相對應。除此之外，myid文件必須存放在每個zookeeper實例的data目錄下，對應本次安裝應該位於 /usr/local/zookeeper/data/zookeeper1，2,3 目錄下，執行如下命令進行配置：
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.啟動zookeeper服務器
使用如下命令啟動zookeeper集群:
cd /usr/local/zookeeper/bin/
./zkServer.sh start ../conf/zoo1.cfg
./zkServer.sh start ../conf/zoo2.cfg
./zkServer.sh start ../conf/zoo3.cfg
啟動后，使用如下命令查看集群狀態:
cd /usr/local/zookeeper/bin/
./zkServer.sh status ../conf/zoo1.cfg ./zkServer.sh status ../conf/zoo2.cfg ./zkServer.sh status ../conf/zoo3.cfg
回顯信息如下：
可以看到有兩個follower節點，一個leader節點。
三、安裝與配置kafka集群
下載地址：http://kafka.apache.org/downloads.html
1.數據目錄和配置文件創建
目前最新版本是2.2.0,本次下載2.1.1版本的安裝包，然后上傳壓縮包到服務器指定目錄，本次上傳目錄為 /software ，然后執行以下命令進行安裝:
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
通過執行上面的命令，我們在 /usr/local/kafka/logs 文件夾中創建了 kafka1,kafka2,kafka3 三個文件夾用於存放三個kafka實例的數據，同時將 /usr/local/kafka/config/ 文件夾下的 server.properties 重命名為 server1.properties 用於配置kafka的第一個實例。
2.配置屬性文件
接下來配置server1.properties 文件，主要配置參數如下：
broker.id=1 ：設置kafka broker的id，本次分別為1,2,3；
delete.topic.enable=true :開啟刪除topic的開關；
listeners=PLAINTEXT://192.168.184.128:9092 ：設置kafka的監聽地址和端口號，本次分別設置為9092,9093,9094；
log.dirs=/usr/local/kafka/logs/kafka1 ：設置kafka日志數據存儲路徑；
zookeeper.connect=192.168.184.128:2181,192.168.184.128:2182,192.168.184.128:2183 ：設置kafka連接的zookeeper訪問地址,集群環境需要配置所有zookeeper的訪問地址;
unclean.leader.election.enable=false ：為true則代表允許選用非isr列表的副本作為leader，那么此時就意味着數據可能丟失，為false的話，則表示不允許，直接拋出NoReplicaOnlineException異常，造成leader副本選舉失敗。
zookeeper.connection.timeout.ms=6000 ：設置連接zookeeper服務器超時時間為6秒。
配置完成后，復制 server1.properties 兩份分別用於配置kafka的第二個，第三個節點：
cd /usr/local/kafka/config/
cp server1.properties server2.properties
cp server1.properties server3.properties
修改修改其中的 broker.id 以及 listeners 、 log.dirs 的配置為第二個，第三個節點的配置，最終各個配置文件配置如下:
server1.properties配置：
server2.properties配置：
server3.properties配置:
3.啟動kafka
通過如下命令啟動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進程:jps |grep -i kafka
說明kafak啟動正常，至此kafka集群搭建完成。本次使用一台服務器作為演示，如果需要在多個服務器上配置集群，配置方法和以上類似，只是不需要像上面那樣配置多個數據目錄和配置文件，每台服務器的配置保持相同，並且注意在防火牆配置端口號即可。
最后，如果需要遠程訪問kafka集群，則需要在防火牆中開通9092、9093、9094端口的訪問權限。
四、測試
1.topic創建與刪除
首先創建一個測試topic，名為testTopic，為了充分利用3個實例（服務器節點），創建3個分區，每個分區都分配3個副本，命令如下：
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創建成功。執行如下命令進行驗證並查看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刪除，使用如下命令進行刪除:
./kafka-topics.sh --zookeeper 192.168.184.128:2181 192.168.184.128:2182 192.168.184.128:2183 --delete --topic testTopic
執行該條命令后，回顯信息如下：
可以看到，testTopic已經被標記為刪除，同時第二行提示表明當配置了 delete.topic.enable 屬性為 true 的時候topic才會刪除，否則將不會被刪除，本次安裝的時候該屬性設置的值為 true 。
2.測試消息發送與消費
首先使用第一步topic創建命令，先創建testTopic這個topic，然后進行消息發送與消費測試。
控制台測試消息發送與消費需要使用kafka的安裝目錄 /usr/local/kafka/bin 下的 kafka-console-producer.sh 來發送消息，使用 kafka-console-consumer.sh 來消費消息。因此本次打開兩個控制台，一個用於執行 kafka-console-producer.sh來發送消息，另一個用於執行 kafka-console-consumer.sh 來消費消息。
消息發送端命令:
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
當發送端和接收端都登錄后，在發送端輸入需要發送的消息並回車，在接收端可以看到剛才發送的消息：
發送端：
接收端：
以上就是簡單地生產消息與消費消息的測試，在測試消費消息的時候時候，命令里邊加了個參數 --from-beginning 表示接收該topic從創建開始的所有消息。
3.生產者吞吐量測試
對於任何一個消息引擎而言，吞吐量是一個至關重要的性能指標。對於Kafka而言，它的吞吐量指每秒能夠處理的消息數或者字節數。kafka為了提高吞吐量，采用追加寫入方式將消息寫入操作系統的頁緩存，讀取的時候從頁緩存讀取，因此它不直接參與物理I/O操作，同時使用以sendfile為代表的零拷貝技術進行數據傳輸提高效率。
kafka提供了 kafka-producer-perf-test.sh 腳本用於測試生產者吞吐量，使用如下命令啟動測試：
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
以上回顯信息表明這台服務器上每個producer每秒能發送6518個消息，平均吞吐量是1.24MB/s，平均延遲2.035秒，最大延遲3.205秒，平均有50%的消息發送需要2.257秒，95%的消息發送需要3.076秒，99%的消息發送需要3.171秒，99.9%的消息發送需要3.205秒。
4.消費者吞吐量測試
與生產者吞吐量測試類似，kafka提供了 kafka-consumer-perf-test.sh 腳本用於消費者吞吐量測試，可以執行以下命令進行測試：
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吞吐量，結果表明該consumer在1秒總共消費了9.5366MB消息。
以上就是kafka集群的搭建以及測試，如有錯誤之處，煩請指正。
參考資料：《Apache kafka實戰》