1 簡(jiǎn)介
??kafka 依賴(lài)zookeeper，zookeeper 簡(jiǎn)稱(chēng)ZK晕鹊，集群官方推薦是奇數(shù)個(gè)server溅话，奇數(shù)個(gè)最少3個(gè)飞几，這樣壞一個(gè)還有兩個(gè)，可以正常選舉躁锁。（官方解釋?zhuān)簩?duì)于復(fù)制模式，需要最少三個(gè)服務(wù)器搜立，并且強(qiáng)烈建議您有奇數(shù)個(gè)服務(wù)器啄踊。如果只有兩個(gè)服務(wù)器颠通，那么蒜哀，如果有一個(gè)服務(wù)器失敗吏砂，則沒(méi)有足夠的機(jī)器形成多數(shù)法定人數(shù)狐血。兩個(gè)服務(wù)器本質(zhì)上不如單個(gè)服務(wù)器穩(wěn)定匈织，因?yàn)橛袃蓚€(gè)單點(diǎn)故障缀匕。）
注意事項(xiàng)：
1、 kafka的java環(huán)境要統(tǒng)一阔加，小版本號(hào)要一致胜榔。
2夭织、 如果安裝kafka集群的服務(wù)器主機(jī)名不是localhost尊惰，需要在/etc/hosts文件里填寫(xiě)集群主機(jī)名和IP地址
2 搭建集群并配置集群
2.1 搭建基礎(chǔ)環(huán)境
由于zookeeper要依賴(lài)java環(huán)境，并且之后的Elasticsearch也需要依賴(lài)java所以首先需要搭建java1.8的基礎(chǔ)環(huán)境戴卜。JDK版本為8u172投剥。
2.1.1 下載jdk1.8：
下載地址：
2.1.2 上傳到服務(wù)器并解壓
[root@kafka11 opt]#tar –zxvf jdk-8u172-linux-x64.tar.gz
2.1.3 修改環(huán)境變量
[root@kafka11 ~]# vim /etc/profile
#jdk1.8
export JAVA_HOME=/opt/jdk1.8.0_172
export JAVA_BIN=/opt/jdk1.8.0_172/bin
export PATH=$PATH:$JAVA_HOME/bin
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME JAVA_BIN PATH CLASSPATH
2.1.4 刷新配置文件
[root@kafka11 ~]# source /etc/profile
2.2 搭建zookeeper集群
版本為：zookeeper3.4.12江锨，官網(wǎng)下地址：
https://archive.apache.org/dist/zookeeper/
2.2.1 配置zookeeper
配置文件模板在$zookeeperhome/conf/zoo_sample.cfg拌消。復(fù)制此模版并且改名為zoo.cfg
[root@kafka11 conf]# cp zoo_sample.cfg zoo.cfg
[root@kafka11 conf]# vim zoo.cfg
lientCnxns=200
tickTime=2000
initLimit=5
syncLimit=2
dataDir=/data/zookeeper/data ##注意此目錄需要手動(dòng)創(chuàng)建
dataLogDir=/data/zookeeper/logs ##此目錄需要手動(dòng)創(chuàng)建
clientPort=2181
server.1=10.10.4.11:2888:3888
server.2=10.10.4.12:2888:3888
server.3=10.10.4.13:2888:3888
2.2.2 創(chuàng)建目錄及myid
[root@kafka11 data]# mkdir –p /data/zookeeper/data
[root@kafka11 data]# mkdir –p /data/zookeeper/logs
[root@kafka11 ~]# cd /data/zookeeper/data/
[root@kafka11 data]# echo 1 > myid ##此次注意鹦筹，要和配置文件里的server.*的*對(duì)應(yīng)
2.2.3 其他節(jié)點(diǎn)配置
注意：myid一定要注意，要和配置文件的server后邊對(duì)應(yīng)的數(shù)字相同相同
2.2.4 Zookeeper命令
開(kāi)啟：[root@kafka12 ~]# /data/zookeeper-3.4.12/bin/zkServer.sh start
停止：[root@kafka12 ~]# /data/zookeeper-3.4.12/bin/zkServer.sh stop
查看狀態(tài)：[root@kafka12 ~]# /data/zookeeper-3.4.12/bin/zkServer.sh status
使用jps命令查看zookeeper進(jìn)程
#執(zhí)行命令jps
20348 Jps
4233 QuorumPeerMain
如果沒(méi)有jps命令可以看端口或者進(jìn)程
2.3 搭建kafka集群
2.3.1 創(chuàng)建目錄并下載安裝軟件
下載地址：版本號(hào)為2.11-0.11.0.3
https://archive.apache.org/dist/kafka/0.11.0.3/kafka_2.11-0.11.0.3.tgz
2.3.2 配置kafka（我的配置文件）
[root@kafka11 config]# cat server.properties
#每個(gè)server需要單獨(dú)配置broker id虚青，如果不配置系統(tǒng)會(huì)自動(dòng)配置。
broker.id=1 ##集群其他服務(wù)器需要改動(dòng)
delete.topic.enable=true
port=9092
host.name=10.10.4.11 ##集群其他服務(wù)器需要改動(dòng)
listeners=PLAINTEXT://:9092
advertised.host.name=10.10.4.11 ##集群其他服務(wù)器需要改動(dòng)
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/logs ##此目錄要先創(chuàng)建达传，不會(huì)自動(dòng)創(chuàng)建。
num.partitions=3 ##分區(qū)個(gè)數(shù)
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.10.4.11:2181,10.10.4.12:2181,10.10.4.13:2181
zookeeper.connection.timeout.ms=6000
session.timeout.ms=10s
max.poll.records=100
max.poll.interval.ms=600000
group.initial.rebalance.delay.ms=0
2.3.3 創(chuàng)建目錄
[root@kafka11 config]# mkdir -p /data/kafka/logs
2.3.4 啟動(dòng)kafka集群
[root@kafka11 config]# cd /data/kafka_2.11-0.11.0.3/bin/
[root@kafka11 bin]# ./kafka-server-start.sh -daemon ../config/server.properties
-daemon參數(shù)是kafka后臺(tái)運(yùn)行的參數(shù)欲主。
2.3.5 檢查是否啟動(dòng)
#執(zhí)行命令jps
20348 Jps
4233 QuorumPeerMain
18991 Kafka
如果沒(méi)有jps命令可以看端口或者進(jìn)程
3 Kafka相關(guān)命令
1、kafka啟動(dòng)命令：
./kafka-server-start.sh -daemon ../config/server.properties
2昧互、kafka查看topic命令：
./kafka-topics.sh --zookeeper 10.10.4.11:2181 --list
3渐逃、查看指定topic的詳細(xì)：
./kafka-topics.sh --zookeeper 10.10.4.11:2181 --topic test_property --describe
4赊堪、創(chuàng)建topic：
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-ref-7
--replication-factor：指定副本個(gè)數(shù)
--partitions：指定分區(qū)個(gè)數(shù)
--topic：指定topic名
5遵绰、寫(xiě)入數(shù)據(jù)：
./kafka-producer-perf-test.sh --num-records 10000000 --topic test-ref-9 --record-size 500 --throughput 100000 --producer-props bootstrap.servers=10.10.4.11:9092,10.10.4.12:9092,10.10.4.13:9092
--num-records：記錄的條數(shù)
--topic：指定topic的名字
--record-size：一條記錄大小
--throughput：吞吐大小（自己感覺(jué)無(wú)用拳喻，只是彈屏的多少）
--producer-props bootstrap.servers=10.10.4.11:9092,10.10.4.12:9092,10.10.4.13:9092：指定kafka集群
6陋葡、消費(fèi)數(shù)據(jù)：
./kafka-consumer-perf-test.sh --messages 10000000 --threads 3 --zookeeper localhost:2181 --num-fetch-threads 3 --topic test-ref-8
--messages：指定消費(fèi)條目數(shù)
--threads：指定線(xiàn)程數(shù)
--num-fetch-threads 3：指定消費(fèi)人數(shù)
4 kafka監(jiān)控工具kafka-manager1.3.3.18安裝使用
4.1 環(huán)境要求
Kafka 0.8.. or 0.9.. or 0.10.. or 0.11..
Java 8+
sbt0.13.*
4.2 安裝部署kafka-manager
4.2.1 安裝sbt
[root@kafka11 ~]# curl https://bintray.com/sbt/rpm/rpm > bintray-sbt-rpm.repo
[root@kafka11 ~]# mv bintray-sbt-rpm.repo /etc/yum.repos.d/
[root@kafka11 ~]# yum install sbt –y
4.2.2 下載
由于下載編譯網(wǎng)站國(guó)內(nèi)訪(fǎng)問(wèn)效果不佳在網(wǎng)上找了個(gè)編譯好的1.3.3.18版本的可以直接使用
鏈接：https://pan.baidu.com/s/1XRXFEmpS4f7KsT5zaaXfTw 提取碼：pb5g
4.2.3 安裝配置
unzip kafka-manager-1.3.3.18.zip
在解壓后的conf目錄中打開(kāi) application.conf文件陨溅，修改其中的zookeeper配置信息：
kafka-manager.zkhosts="hadoop4:2181,hadoop5:2181,hadoop6:2181"
##用逗號(hào)隔開(kāi)
4.2.4 指定端口啟動(dòng)
nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port 9001 &
或者修改配置文件
添加http.port=9001
4.3 使用manager
1偿渡、添加一個(gè)集群
點(diǎn)擊【Cluster】>【Add Cluster】打開(kāi)如下添加集群的配置界面：
輸入集群的名字（如Kafka-1）和 Zookeeper 服務(wù)器地址（如10.10.4.11:2181）。選擇最接近的Kafka版本（如0.10.1.0）留攒。
注意：如果沒(méi)有在 Kafka 中配置過(guò) JMX_PORT，千萬(wàn)不要選擇第一個(gè)復(fù)選框。
Enable JMX Polling。如果選擇了該復(fù)選框腔剂，Kafka-manager 可能會(huì)無(wú)法啟動(dòng)登渣。
2、新建主題
點(diǎn)擊【Topic】>【Create】可以方便的創(chuàng)建并配置主題。如下顯示。
5 Kafka監(jiān)控工具KafkaOffsetMonitor配置及使用
5.1 安裝及配置
1王财、正常直接去官網(wǎng)下載最新的jar包，上傳到某服務(wù)器上改览，可以直接運(yùn)行就可以運(yùn)行起來(lái)了：https://github.com/quantifind/KafkaOffsetMonitor/releases
但是在使用過(guò)程中有可能會(huì)發(fā)現(xiàn)頁(yè)面反應(yīng)緩慢或者無(wú)法顯示相應(yīng)內(nèi)容的情況今妄。據(jù)說(shuō)這是由于jar包中的某些js等文件需要連接到網(wǎng)絡(luò)瞻离，或者需要翻墻導(dǎo)致的推励。網(wǎng)上找的一個(gè)修改版的KafkaOffsetMonitor對(duì)應(yīng)jar包，可以完全在本地運(yùn)行，經(jīng)過(guò)測(cè)試效果不錯(cuò)。我已經(jīng)上傳到我的網(wǎng)盤(pán)：鏈接：https://pan.baidu.com/s/1sCZQl6K0zcPGeS54MTVoGw 提取碼：njsd
2、上傳解壓
2、在解壓目錄里創(chuàng)建一個(gè)新的腳本
#!/bin/bash
java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \
com.quantifind.kafka.offsetapp.OffsetGetterWeb \
--zk 10.10.4.11:2181,10.10.4.12:2181,10.10.4.13:2181 \
--port 8088 \
--refresh 5.minutes \
--retain 1.day
--zk localhost:2181——指的是zookeeper的IP和端口號(hào)
--port 8088——指的是KafkaOffsetMonitor訪(fǎng)問(wèn)的端口號(hào)，即監(jiān)控kafka的端口號(hào)（當(dāng)前系統(tǒng)沒(méi)有用到的端口號(hào)）
--refresh 10.seconds——10秒一刷新
--retain 1.days——http://localhost:8089頁(yè)面保持1天
3钻哩、然后運(yùn)行腳本
之后會(huì)有提示讓你訪(fǎng)問(wèn)：IP:8088
5.2 應(yīng)用（只有在消費(fèi)的時(shí)候可以看到數(shù)據(jù)）
1珊肃、 剛進(jìn)入界面
2、有一個(gè)Visualizations選項(xiàng)卡，點(diǎn)擊其中的Cluster Overview可以查看當(dāng)前Kafka集群的Broker情況
