~斯汀的IT世界~

集群規劃如下:

Linux01	Linux02	Linux03
zookeeper	zookeeper	zookeeper
kafka	kafka	kafka

 

1.zookeeper 安裝

[root@linux01]# yum install java-11-openjdk-devel

[root@linux01]# mkdir /tmp/zookeeper

[root@linux01]# vim /etc/systemd/system/zookeeper.service
修改為以下內容:
[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target
[Service]
Type=simple
User=root
ExecStart=/usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties
ExecStop=/usr/local/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal
[Install]
WantedBy=multi-user.target

[root@linux01]# vim /usr/local/kafka/config/zookeeper.properties
修改為以下內容:
dataDir=/tmp/zookeeper
clientPort=2181
maxClientCnxns=0
admin.enableServer=false
tickTime=2000
initLimit=5
syncLimit=2
server.1=kafka01:2888:3888
server.2=kafka02:2888:3888
server.3=kafka03:2888:3888

[root@linux01]# echo “1” > /tmp/zookeeper/myid

在 Linux02 與 Linux03 主機中執行上述的步驟
注意：myid裡的設定值不可以重複

[root@linux01 kafka]# systemctl start zookeeper

[root@linux01 kafka]# systemctl enable zookeeper

[root@linux01 kafka]# jps
15929 Jps
12123 QuorumPeerMain
13486 Kafka

[root@linux01 kafka]# bin/zookeeper-shell.sh 127.0.0.1:2181
ls [path]：
ls /
[zookeeper, brokers]
get [path]：
#查看broker的信息
get /brokers/ids/0

2.Kafka 集群部署

2.2 Kafka 集群部署

http://ftp.mirror.tw/pub/apache/kafka/2.5.0/kafka_2.12-2.5.0.tgz

 

1.解壓縮安裝檔

[root@linux01]# tar -zxvf kafka_2.12-2.5.0.tgz -C /usr/local

2.建立軟連結

[root@linux01]# cd /usr/local/kafka

[root@linux01 kafka]# ln -s kafka_2.12-2.5.0 kafka

3.建立 logs 資料夾

[root@linux01 kafka]# mkdir /tmp/kafka-logs

4.配置設定檔

[root@linux01 kafka]# cd config

[root@linux01 config]# vim server.properties​
修改為以下內容:
broker.id=1
delete.topic.enable=true
log.dirs=/tmp/kafka-logs
zookeeper.connect=kafka01:2181,kafka02:2181,kafka03:2181

設定環境變數 (不一定要用)

[root@linux01 kafka]# vim /etc/profile
# KAFKA_HOME
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
[root@linux01 kafka]# source /etc/profile

6.    將安裝檔複製到Linux02 與 Linux03主機中

7.    在 Linux02 與 Linux03 主機中執行上述的步驟
注意：設定檔中的broker.id 不可以重複
8.    啟動集群
[root@linux01 kafka]# vim /etc/systemd/system/kafka.service

[Unit]
Requires=zookeeper.service
After=zookeeper.service
[Service]
Type=simple
User=root
ExecStart=/bin/sh -c '/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties > /var/log/kafka.log 2>&1'
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal
[Install]
WantedBy=multi-user.target

[root@linux01 kafka]# systemctl start kafka
[root@linux01 kafka]# systemctl enable kafka

9.    關閉集群
[root@linux01 kafka]# systemctl stop kafka

2.3 Kafka 指令操作

1.    查詢目前所有的topic

舊版
[root@linux01 kafka]# bin/kafka-topics.sh --list --zookeeper localhost:2181

新版
[root@linux01 kafka]# bin/kafka-topics.sh --list --bootstrap-server kafka01:9092

2.    建立 topic

舊版
[root@linux01 kafka]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 3 --topic test

新版
[root@linux01 kafka]# bin/kafka-topics.sh --create --bootstrap-server kafka01:9092 --replication-factor 2 --partitions 3 --topic tony

3.    查看某個topic的資訊

[root@kafka01 kafka]# bin/kafka-topics.sh --describe --bootstrap-server kafka01:9092 --topic test

4.    刪除 topic (需要在server.properties中設定delete.topic.enable=true)

[root@linux01 kafka]# bin/kafka-topics.sh --delete --bootstrap-server kafka01:9092 --topic tony

5.    發送訊息
[root@linux01 kafka]# bin/kafka-console-producer.sh --broker-list kafka01:9092 --topic test

6.    消費訊息
[root@linux02 kafka]# bin/kafka-console-consumer.sh --bootstrap-server kafka02:9092 --topic test --from-beginning

補充：
[root@linux01 kafka]# strings 00000000000000000000.log

7.    Trigger Preferred Leader Election

[root@linux01 kafka]# kafka-leader-election.sh --bootstrap-server kafka01:9092 --topic test --partition 0 --election-type preferred

Successfully completed leader election (PREFERRED) for partitions test-0