前言

注意：Kafka是v2.8之前的版本，从v2.8版本开始不再需要借助ZooKeeper

kafka v2.8开始弃用ZooKeeper

kafka v3.0开始弃用Java 8 和 Scala 2.12

该文章用的Kakfa版本是v2.1版本。如果想参考v2.8之后的版本，点击【Kafka的v3版本在Linux下载安装及部署】

前期准备工作：
Kafka的安装及使用需要用到ZooKeeper，所以需要提前安装搭建好ZooKeeper
Zookeeper在Linux下载安装及部署

kafka的安装及部署

官网下载 kafka.tgz 包

我这里安装的版本：kafka_2.12-2.2.0.tgz

Kafka官网下载地址：Apache Kafka官网

上传服务器

将下载好的kafka_2.12-2.2.0.tgz包丢到服务器上任意一个你能找到的地方

创建文件夹

在/usr/local路径下创建一个文件夹kafka

1
mkdir /usr/local/kafka

解压

解压压缩包到/usr/local/kafka目录下

1
tar -zxvf kafka_2.12-2.2.0.tgz -C /usr/local/kafka

新建kafka-logs

在 /usr/local/kafka/kafka_2.12-2.2.0新建一个文件夹kafka-logs

1
mkdir /usr/local/kafka/kafka_2.12-2.2.0/kafka-logs

修改kafka的配置文件

在kafka主目录下 bin/config或bin/config/kraft文件夹中找到server.properties并进行修改

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

vim bin/config/server.properties

注意：kafka在启动服务之前，在server.properties文件中要设定3个参数： broker.id、log.dirs、zookeeper.connect

1
2
3
4
5
6
7
8
9
10
11
12
13
14
broker.id=0
 
log.dirs=/usr/local/kafka/kafka_2.12-2.2.0/kafka-logs
 
zookeeper.connect=localhost:2181
 
delete.topic.enable=true
 
advertised.listeners=PLAINTEXT://localhost:9092
delete.topic.enable=true  ：这段代码会对以后删除kafka中的topic有影响，这段代码在文件尾部添加上即可

listeners=PLAINTEXT://:9092 ：这个命令也很重要，需要记住(这个命令在文章里先不做分析)

advertised.listeners=PLAINTEXT://自己的主机ip地址:9092：这个localhost我用的是主机ip地址

配置kafka环境变量

配置kafka环境变量，配置zookeeper 和 kafka的全局命令

刷新Kafka生效配置

方法一：修改profile文件

我使用的该方法，也推荐用这一种，两钟区别需自行查阅相关资料

1
vim /etc/profile

直接在最下面添加下面这些配置

1
2
export KAFKA_HOME=/usr/local/kafka/kafka_2.12-2.2.0
export PATH=$KAFKA_HOME/bin:$PATH

使配置生效

1
source /etc/profile

方法二：修改.bashrc 文件

输入命令修改环境变量

1
vim ~/.bashrc

直接在最下面添加下面这些配置

1
2
export KAFKA_HOME=/usr/local/kafka/kafka_2.12-2.2.0
export PATH=KAFKA_HOME/bin:$PATH

使配置生效

1
source ~/.bashrc

验证kafka是否已经生效

验证环境变量是否配置成功

1
echo $KAFKA_HOME

验证输入kafka- 后按teb键是否会自动补全

1
kafka-

kafka的使用

启动Zookeeper服务

Kafka用到了Zookeeper，先开启zookeeper，再开启Kafka （依次开启）

下面用一个单实例的Zookkeeper服务，可以在命令结尾处加个&符号，这样就可以启动后离开控制台（进程守护），启动kafak命令同理

启动Zookeeper服务，在 kafka 的根目录下使用命令

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/zookeeper-server-start.sh config/zookeeper.properties &

启动kafka

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/kafka-server-start.sh config/server.properties &

创建一个topic

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

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

查看kafka的topic情况

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

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

疑问：—zookeeper localhost:2181”是什么意思？？？

回答：【—zookeeper】是kafka连接zookeeper的url，和server.properties文件中的配置项 zookeeper.connect=localhost:2181 一致

描述topic，查看topic的详细信息

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic1

生产消费

生产消息，生产者客户端命令，在 kafka 的根目录下使用命令

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic1

这里的 --broker-list localhost:9092 是

进入 生产消息的编辑模式，进行编辑并发送


1
2
>hello,world
>this is my kafka

消费消息，消费者客户端命令，在 kafka 的根目录下使用命令


1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic1 --from-beginning

消费消息如下：


1
2
hello,world
this is my kafka

删除 topic

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topic1

或者

1
./bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper 127.0.0.1:2181 --topic topic1

删除topic 补充 ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓ ↓

问题：
对 topic1 进行删除操作后，再次查看topic的时候，topic1 没有直接删除，而且后面出现了 “marked for deletion” ，只是将 topic1 标记了删除

解决：（通过查阅各种解决方案，整理了下面这种解决方案）

如果没有在kafka/config/server.properties配置：delete.topic.enable=true时，执行topic的删除命令只是把topic标记为marked for deletion，并不是真正的删除，如果此时想彻底删除，就需要登录zookeeper客户端进行删除

启动zookeeper客户端，在zookeeper根目录下使用命令：
cd 

1
2
3
cd /usr/local/zookeeper/apache-zookeeper-3.5.5

./bin/zkCli.sh

找到topic所在的目录，在zookeeper客户端输入这个命令

1
ls /brokers/topics

删除需要彻底删除的 topic1，在zookeeper客户端输入这个命令

1
deleteall /brokers/topics/topic1

rmr 命令已经被废弃了，可以执行 deleteall 命令

可以再次查看确认一下，在zookeeper客户端输入这个命令

1
ls /brokers/topics

在 config/server.properties 中找到log.dirs，

删除log.dirs指定的文件目录，然后重新启动

验证是否已经删除

通过下面命令验证，发现已经删除了

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

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

关闭服务

先关闭Kafka，在 kafka 的根目录下使用命令

1
2
3
cd /usr/local/kafka/kafka_2.12-2.2.0

./bin/kafka-server-stop.sh

再关闭Zookeeper，在 Zookeeper 的根目录下使用命令
1
2
3
cd /usr/local/zookeeper/apache-zookeeper-3.5.5

./bin/zkServer.sh stop

验证一下是否关闭

jps:查看当前运行的进程
jps -m:查看当前运行进程的详细信息

1
jps

有问题，有错误及时提出来，共同进步~~

注意：在实际项目中使用，记住hostname需要配置

kafka的相关案例

如何轻松在SpringBoot中正确配置并运行Kafka？​​​​​​​