集群環(huán)境

kafka+zk

我這里并沒有設置專業(yè)的存儲設備，大家用到生產(chǎn)中一定要把數(shù)據(jù)存儲到專業(yè)的存儲設備或者帶有冗余設備的磁盤上

###部署JDK

tar xf jdk-8u161-linux-x64.tar.gz  -C /usr/local/
cat << EOF >> /etc/profile
#################JAVA#################
export JAVA_HOME=/usr/local/jdk1.8.0_161
export JRE_HOME=$JAVA_HOME/jre
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
export PATH=$JAVA_HOME/bin:$JRE_HOME/bin:$PATH
EOF

source /etc/profile
java -version

###部署ZooKeeper ZooKeeper官方站點：https://zookeeper.apache.org/ Kafka使用ZooKeeper來管理，因此需要安裝ZooKeeper，并且要先啟動ZooKeeper

wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

解壓安裝

tar xf zookeeper-3.4.14.tar.gz  -C /usr/local/
cp -rf /usr/local/zookeeper-3.4.14/conf/zoo_sample.cfg /usr/local/zookeeper-3.4.14/conf/zoo.cfg

修改ZooKeeper配置 三臺設備保持一致，配置文件詳解請看這篇文章內(nèi)的配置文件詳解部分：https://abcops.cn/zookeeper-single-deployment/

cat << EOF > /usr/local/zookeeper-3.4.14/conf/zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zk_data
dataLogDir=/usr/local/zookeeper-3.4.14/logs
clientPort=2181
maxClientCnxns=60
autopurge.snapRetainCount=3
autopurge.purgeInterval=1
server.1=172.17.0.2:2888:3888
server.2=172.17.0.3:2888:3888
server.3=172.17.0.4:2888:3888
EOF

創(chuàng)建ZooKeeper數(shù)據(jù)存儲目錄及日志目錄

mkdir -p /data/zk_data
mkdir /usr/local/zookeeper-3.4.14/logs

創(chuàng)建ServerID標識

在ZooKeeper集群中除配置文件外，還需要配置一個myid文件，這個文件需要存放在配置文件中dataDir配置項所指定的數(shù)據(jù)位置，要根據(jù)集群中的節(jié)點創(chuàng)建不用的文件。我們要根據(jù)ServerID標示來創(chuàng)建相應的文件

Kafka_node1

[root@kafka_node1 /]# echo '1' > /data/zk_data/myid

Kafka_node2

[root@kafka_node2 /]# echo '2' > /data/zk_data/myid

Kafka_node3

[root@kafka_node3 /]# echo '3' > /data/zk_data/myid

啟動ZK集群并查看 三臺ZK節(jié)點全部啟動

/usr/local/zookeeper-3.4.14/bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.4.14/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

查看ZK端口監(jiān)聽情況 losf -i:命令來查看端口信息 leader：如果此設備是leader，那么使用losf查看到的連接會是與集群內(nèi)所有的follower的連接 follower：如果此設備是follower，那么使用losf查看到的連接將只會與ZK集群中的leader連接

Kafka_node1 node1只有一個連接是和172.17.0.3建立的，可以表明此節(jié)點為follower節(jié)點

[root@kafka_node1 /]# lsof -i:2888
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    15620 root   30u  IPv4  71449      0t0  TCP kafka_node1:42424->172.17.0.3:spcsdlobby (ESTABLISHED)

Kafka_node2 node2是與集群內(nèi)的其它兩臺機器所連接，可以表明此節(jié)點為leader節(jié)點

[root@kafka_node2 /]# lsof -i:2888
COMMAND PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java     91 root   30u  IPv4  65243      0t0  TCP kafka_node2:spcsdlobby (LISTEN)
java     91 root   31u  IPv4  69183      0t0  TCP kafka_node2:spcsdlobby->172.17.0.2:42424 (ESTABLISHED)
java     91 root   33u  IPv4  69192      0t0  TCP kafka_node2:spcsdlobby->172.17.0.4:49420 (ESTABLISHED)

Kafka_node3 node3與node1一樣為follower節(jié)點

[root@kafka_node3 /]# lsof -i:2888
COMMAND PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java     91 root   31u  IPv4  64087      0t0  TCP kafka_node3:49420->172.17.0.3:spcsdlobby (ESTABLISHED)

到此ZK集群搭建完畢，如需了解ZK更多的管理操作，請看：https://abcops.cn/zookeeper-single-deployment/

###部署Kafka 下載安裝

wget http://mirror.bit.edu.cn/apache/kafka/2.2.1/kafka_2.12-2.2.1.tgz
tar xf kafka_2.12-2.2.1.tgz -C /usr/local/

修改Kafka配置

關(guān)于Kafka配置詳解請參考：https://abcops.cn/kafka-config-file/

原文件備份

cp -rf /usr/local/kafka_2.12-2.2.1/config/server.properties /usr/local/kafka_2.12-2.2.1/config/server.properties.default

配置修改

cat << EOF > /usr/local/kafka_2.12-2.2.1/config/server.properties
broker.id=1                                                     #Kafka_node2節(jié)點修改為2，3修改為3
listeners=PLAINTEXT://172.17.0.2:9092                           #修改Kafka_node的IP地址為各自node本地地址
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/
num.partitions=1
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=72
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=172.17.0.2:2181,172.17.0.3:2181,172.17.0.4:2181
delete.topic.enable=true
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=3000
EOF

啟動Kafka 三臺Kafka_node全部啟動

/usr/local/kafka_2.12-2.2.1/bin/kafka-server-start.sh -daemon /usr/local/kafka_2.12-2.2.1/config/server.properties

查看進程連接

jps
15620 QuorumPeerMain
16057 Jps
15945 Kafka

下面通過過濾端口號可以看到
ZooKeeper監(jiān)控本地地址TCP端口2181，可以ZooKeeper看到2181后面對應的還有一個端口號為43918
Kafka監(jiān)控本地地址TCP端口9092，可以看到Kafka9092后面也有對應的一個端口號55568
netstat -anplt | egrep "(2181|9092)"
tcp        0      0 172.17.0.2:9092         0.0.0.0:*               LISTEN      15945/java          
tcp        0      0 0.0.0.0:2181            0.0.0.0:*               LISTEN      15620/java          
tcp        0      0 172.17.0.2:45674        172.17.0.4:9092         ESTABLISHED 15945/java          
tcp        0      0 172.17.0.2:55568        172.17.0.2:9092         ESTABLISHED 15945/java          
tcp        0      0 172.17.0.2:9092         172.17.0.2:55568        ESTABLISHED 15945/java          
tcp        0      0 172.17.0.2:43094        172.17.0.3:9092         ESTABLISHED 15945/java          
tcp        0      0 172.17.0.2:2181         172.17.0.2:43918        ESTABLISHED 15620/java          
tcp        0      0 172.17.0.2:43918        172.17.0.2:2181         ESTABLISHED 15945/java   

查看Kafka監(jiān)聽狀態(tài) Kafka_node1 可以看到node1節(jié)點同時與node2及node3建立了連接，可以看到node1節(jié)點為主導者

[root@kafka_node1 /]# lsof -i:9092
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    15945 root  106u  IPv4  64487      0t0  TCP kafka_node1:XmlIpcRegSvc (LISTEN)
java    15945 root  122u  IPv4  72859      0t0  TCP kafka_node1:55568->kafka_node1:XmlIpcRegSvc (ESTABLISHED)
java    15945 root  123u  IPv4  71815      0t0  TCP kafka_node1:XmlIpcRegSvc->kafka_node1:55568 (ESTABLISHED)
java    15945 root  127u  IPv4  72997      0t0  TCP kafka_node1:43094->172.17.0.3:XmlIpcRegSvc (ESTABLISHED)
java    15945 root  131u  IPv4  75769      0t0  TCP kafka_node1:45674->172.17.0.4:XmlIpcRegSvc (ESTABLISHED)

Kafka_node2 node2只與node1建立了連接

[root@kafka_node2 /]# lsof -i:9092
COMMAND PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    415 root  106u  IPv4  75016      0t0  TCP kafka_node2:XmlIpcRegSvc (LISTEN)
java    415 root  116u  IPv4  72998      0t0  TCP kafka_node2:XmlIpcRegSvc->172.17.0.2:43094 (ESTABLISHED)

Kafka_node3 node3與node2一樣和node1建立了連接

9092
COMMAND  PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    1125 root  106u  IPv4  73394      0t0  TCP kafka_node3:XmlIpcRegSvc (LISTEN)
java    1125 root  116u  IPv4  72253      0t0  TCP kafka_node3:XmlIpcRegSvc->172.17.0.2:45674 (ESTABLISHED)

###管理Kafka

接下來我們操作下Kafka，我們會通過kafka-topics.sh新建一個Topic，然后使用kafka-console-producer.sh消息生產(chǎn)腳本來生產(chǎn)消息到Topic中，再由kafka-console-consumer.sh消息消費者消費消息，以及常用的選項介紹。

創(chuàng)建Topic主題

[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --create --bootstrap-server 172.17.0.2:9092,172.17.0.3:9092,172.17.0.4:9092 --replication-factor 3 --partitions 3 --topic kafka_data
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
#上面kafka警告說“在創(chuàng)建Topic時不允許使用"_."之類的符號，由于我們是測試，沒有關(guān)系”

選項解釋：
--create：創(chuàng)建新的Topic
--bootstrap-server：指定要哪臺Kafka服務器上創(chuàng)建Topic，主機加端口，指定的主機地址一定要和配置文件中的listeners一致
--zookeeper：指定要哪臺zookeeper服務器上創(chuàng)建Topic，主機加端口，指定的主機地址一定要和配置文件中的listeners一致
--replication-factor：創(chuàng)建Topic中的每個分區(qū)(partition)中的復制因子數(shù)量，即為Topic的副本數(shù)量，建議和Broker節(jié)點數(shù)量一致，如果復制因子超出Broker節(jié)點將無法創(chuàng)建
--partitions：創(chuàng)建該Topic中的分區(qū)(partition)數(shù)量
--topic：指定Topic名稱

查看已創(chuàng)建的Topic

剛才我們創(chuàng)建了3份Topic，創(chuàng)建時指定了三臺Kafka的地址及端口，所以每天上面都會有我們創(chuàng)建的Topic
[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --list --bootstrap-server 172.17.0.2:9092
kafka_data
[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --list --bootstrap-server 172.17.0.3:9092
kafka_data
[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --list --bootstrap-server 172.17.0.4:9092
kafka_data

生產(chǎn)消息

[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-console-producer.sh --broker-list 172.17.0.2:9092 --topic kafka_data
>Hello Kafka_data
>I'm the 172.17.0.2 Kafka create
>test

參數(shù)解釋：
--broker-list：指定使用哪臺broker來生產(chǎn)消息
--topic：指定要往哪個Topic中生產(chǎn)消息

消費消息 我們在Kafka_node1上的Topic生產(chǎn)消息，在Kafka_node3中消費消息，可以得出，Broker存儲過消息后會同步給集群內(nèi)的其它Broker節(jié)點

[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-console-consumer.sh --bootstrap-server 172.17.0.4:9092 --topic kafka_data --from-beginning 
I'm the 172.17.0.2 Kafka create
test
Hello Kafka_data

查看Topic詳情

[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --describe --bootstrap-server 172.17.0.2:9092 --topic kafka_data
Topic:kafka_data    PartitionCount:3    ReplicationFactor:3    Configs:segment.bytes=1073741824
    Topic: kafka_data    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3
    Topic: kafka_data    Partition: 1    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    Topic: kafka_data    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2

參數(shù)解釋：
Topic:kafka_data：topic名稱
PartitionCount:3：分片數(shù)量
ReplicationFactor:3：Topic副本數(shù)量

刪除Topic

[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --delete --bootstrap-server 172.17.0.2:9092 --topic kafka_data

查看刪除信息
[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --list --bootstrap-server 172.17.0.2:9092
__consumer_offsets
[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --list --bootstrap-server 172.17.0.3:9092
__consumer_offsets
[root@kafka_node1 /]# /usr/local/kafka_2.12-2.2.1/bin/kafka-topics.sh --list --bootstrap-server 172.17.0.4:9092
__consumer_offsets

我們在node1節(jié)點刪除了Topic，三臺節(jié)點會同步更新，所以我們的kafka_data在三臺node上全部刪除

好啦！今天的分享到這里就結(jié)束了，希望大家持續(xù)關(guān)注馬哥教育官網(wǎng)，每天都會有大量優(yōu)質(zhì)內(nèi)容與大家分享！

文章來源于網(wǎng)絡，侵刪！