1、下载kafka二进制文件

https://kafka.apache.org/downloads
wget https://archive.apache.org/dist/kafka/2.5.0/kafka_2.13-2.5.0.tgz

2、解压kafka二进制文件

cd /opt/soft
wget https://archive.apache.org/dist/kafka/2.5.0/kafka_2.13-2.5.0.tgz
tar -xzvf kafka_2.13-2.5.0.tgz
cd kafka_2.13-2.5.0

3、启动zookeeper

kafka需要依赖ZK，安装包中已经自带了一个ZK，也可以改成指定已运行的ZK
如果改成指定的ZK需要修改修改 kafka 安装目录下的 config/server.properties 文件中的 zookeeper.connect ，这里使用自带的ZK

3.1、后台启动ZK

cd /opt/soft/kafka_2.13-2.5.0
nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties >> zookeeper.nohup &

3.2、检查zookeeper是否启动成功

ps -ef|grep zookeeper

4、启动kafka

4.1、修改配置

vim config/server.properties
broker.id=1      # Broker ID启动以后就不能改了
listeners=PLAINTEXT://192.168.122.188:9092   # kafka监听地址

# num.partitions后面增加2行。
# 发送到不存在topic自动创建。允许永久删除topic
num.partitions=1
auto.create.topics.enable=true
delete.topic.enable=true

4.2、后台启动kafka

cd /opt/soft/kafka_2.13-2.5.0
nohup ./bin/kafka-server-start.sh ./config/server.properties & 

4.3、创建Topic

4.3.1、创建一个名为gptest的topic，只有一个副本，一个分区

sh bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic gptest

4.4、查看已经创建的 topic：

sh bin/kafka-topics.sh -list -zookeeper localhost:2181

5、启动Producer

5.1、打开一个窗口，在kafka解压目录下

cd /opt/soft/kafka_2.13-2.5.0
sh bin/kafka-console-producer.sh --broker-list 192.168.122.188:9092 --topic gptest

6、启动Consumer

在一个新的远程窗口中

sh bin/kafka-console-consumer.sh --bootstrap-server 192.168.122.188:9092 --topic gptest --from-beginning

7、Producer窗口发送消息

7.1、输入hello world 回车

[root@master kafka_2.13-2.5.0]# sh bin/kafka-console-producer.sh --broker-list 192.168.122.188:9092 --topic gptest
>
>gptest
>hello world

8、Consumer端可以查看到消息的消费

[root@master kafka_2.13-2.5.0]# sh bin/kafka-console-consumer.sh --bootstrap-server 192.168.122.188:9092 --topic gptest --from-beginning
gptest
hello world

9、删除kafka全部数据步骤：

9.1、停止每台机器上的kafka；

9.2、删除kafka存储目录（server.properties文件log.dirs配置，默认为“/tmp/kafka-logs”）全部topic的数据目录

9.3、删除zookeeper上与kafka相关的znode节点；除了/zookeeper

9.4、重启kafka