下面我们演示一下如何在一台机器上安装部署最新版的kafka（版本是3.2.1） 3节点集群，并且使用 kraft模式，不使用zookeeper。

1. 下载最新版的安装包

（截止2022-08-16）
点击 此处进入到kafka官方网站的下载页面 ，会看到如下页面：

点击上图中的红框处或点击 此处 即可下载

2. 解压安装包并重命名

因为我只有一台机器，需要在一台机器上搭建3节点的集群，所以使用不同的目录来区分不同的节点。
首先解压下载的kafka_2.13-3.2.1.tgz到自己的目录

tar -zxvf kafka_2.13-3.2.1.tgz

得到目录 kafka_2.13-3.2.1 ，这里使用如下命令重命名为 kafka-1，表示第1个几点。

mv kafka_2.13-3.2.1 kafka-1

3. 配置文件目录说明

因为这里我们使用是kraft模式，不使用zookeeper，需要修改 kafka-1/config/kraft 目录下的配置，我们进入该目录：

cd kafka-1/config/kraft

可以看到如下4个文件

README.md -- 这个文件就是kraft模式的使用说明
broker.properties
controller.properties
server.properties -- 这个就是我们需要修改的配置文件

注意：我们查阅 README.md，可以发现 kraft目前还是不推荐在生产上使用，只能用于测试。

4. 修改节点1的配置server.properties

我们在该目录 kafka-1/config/kraft 编辑server.properties

vim server.properties

因server.properties内容较长，下面只写出我们需要修改的内容，如下：

下面是第 1 处的修改，node.id表示节点的id，这里我们设置为 1
# The node id associated with this instance's roles
node.id=1

下面是第 2 处的修改，controller.quorum.voters表示参与投票的节点，我们使用一台机器，用3个端口来区分，格式是 node.id@ip:port
# The connect string for the controller quorum
controller.quorum.voters=1@localhost:7301,2@localhost:7303,3@localhost:7305

下面是第 3 处的修改，PLAINTEXT 表示对外提供服务（客户端生产和消费）的端口，CONTROLLER和上面的controller.quorum.voters 保持一致，用于节点之间信息交互
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:7302,CONTROLLER://localhost:7301

下面是第 4 处的修改，advertised.listeners和上面的PLAINTEXT一致都是7302
# If not set, it uses the value for "listeners".
advertised.listeners=PLAINTEXT://localhost:7302

下面是第 5 处的修改，log.dirs表示日志的目录，也是kafka数据储存的位置。注意需要让 linux 的当前用户对/tmp/目录可读可写
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kraft-combined-logs/1

5. 复制kafka-1模拟3节点

回到 kafka-1 的上级目录，使用如下命令再复制出两个目录，模拟节点2和节点3。

cp -r kafka-1 kafka-2
cp -r kafka-1 kafka-3

6. 修改节点2和节点3的配置

修改节点2的配置，这里还是只写出我们需要修改的5处，如下：

# The node id associated with this instance's roles
node.id=2

# The connect string for the controller quorum
controller.quorum.voters=1@localhost:7301,2@localhost:7303,3@localhost:7305

#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:7304,CONTROLLER://localhost:7303

# If not set, it uses the value for "listeners".
advertised.listeners=PLAINTEXT://localhost:7304

# A comma separated list of directories under which to store log files
log.dirs=/tmp/kraft-combined-logs/2

修改节点3的配置，这里也只写出我们需要修改的5处，如下：

# The node id associated with this instance's roles
node.id=3

# The connect string for the controller quorum
controller.quorum.voters=1@localhost:7301,2@localhost:7303,3@localhost:7305

#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:7306,CONTROLLER://localhost:7305

# If not set, it uses the value for "listeners".
advertised.listeners=PLAINTEXT://localhost:7306

# A comma separated list of directories under which to store log files
log.dirs=/tmp/kraft-combined-logs/3

至此，3个节点的配置文件就全部修改好了。

7. 运行KRaft集群

运行KRaft集群，主要分为三步，这部分在上面提到的 README.md 中都有说明。

7.1 在任意节点生成一个唯一的集群ID

到任意一个节点的目录下，比如 kafka-1 里面运行

cd kafka-1
./bin/kafka-storage.sh random-uuid

会有如下输出

04ofzeqFRgqBWQGtLEqmNQ

这里就生成了一个ID，注意这个ID需要在所有节点中使用。

7.2 格式化存储数据的目录

进入每一个节点的目录下，运行如下命令：

cd kafka-1
./bin/kafka-storage.sh format -t 04ofzeqFRgqBWQGtLEqmNQ -c ./config/kraft/server.properties

cd kafka-2
./bin/kafka-storage.sh format -t 04ofzeqFRgqBWQGtLEqmNQ -c ./config/kraft/server.properties

cd kafka-3
./bin/kafka-storage.sh format -t 04ofzeqFRgqBWQGtLEqmNQ -c ./config/kraft/server.properties

可以看到上面使用的是一个集群ID。

7.3 启动Kafka Server

在每一个节点运行如下启动命令

cd kafka-1
./bin/kafka-server-start.sh ./config/kraft/server.properties

cd kafka-2
./bin/kafka-server-start.sh ./config/kraft/server.properties

cd kafka-3
./bin/kafka-server-start.sh ./config/kraft/server.properties

当然您也可以使用nohup 和 & 后台启动。
这时 kraft模式的kafka 集群就成功启动了。

8. 测试

如下操作都只需在任意一个节点执行，比如kafka-1，先进入到kafka-1的目录：

cd kafka-1

8.1 创建topic

./bin/kafka-topics.sh --create --topic test-topic --partitions 1 --replication-factor 1 --bootstrap-server localhost:7302 

8.2 生产数据

 ./bin/kafka-console-producer.sh --broker-list localhost:7304 --topic test-topic

接着在控制台输入：

> hello, mcdull. 

按回车键即可发送数据。

8.3 消费数据

./bin/kafka-console-consumer.sh --bootstrap-server localhost:7306 --topic test-topic --from-beginning

输出如下：

hello, mcdull.