一、概要

Kafka作为一种高吞吐量的分布式发布订阅消息系统，在消息应用中广泛使用，在Kafka2.8之前，Kafka强依赖zookeeper，这也导致当Zookeeper集群性能发生抖动时，Kafka的性能也会收到很大的影响。2.8版本之后，kafka3.x开始提供KRaft（Kafka Raft，依赖Java 8+ ）模式，开始去除对zookeeper的依赖。最新的3.5版本中，Kafka依然兼容zookeeper Controller，但Kafka Raft元数据模式，已经可以在不依赖zookeeper的情况下独立启动Kafka了。

kraft模式的优势：

1、更简单的部署和管理——通过只安装和管理一个应用程序，Kafka 现在的运营足迹要小得多。这也使得在边缘的小型设备中更容易利用 Kafka；
2、提高可扩展性——KRaft 的恢复时间比 ZooKeeper 快一个数量级。这使我们能够有效地扩展到单个集群中的数百万个分区。ZooKeeper 的有效限制是数万；
3、更有效的元数据传播——基于日志、事件驱动的元数据传播可以提高 Kafka 的许多核心功能的性能。

资料链接：官网kraft

二、拓扑

2.1、早期拓扑

2.2、controller拓扑

三、部署配置

wget 
tar -zxvf kafka_2.13-3.4.0.tgz -C /opt/
mv /opt/kafka_2.13-3.4.0. /opt/kafka
chown kafka:kafka -R /opt/kafka
cd /opt/kafka/
mkdir data
vim /opt/kafka/config/kraftserver.properties  //如下所示

# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller

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

# The connect string for the controller quorum 集群选举控制器配置
controller.quorum.voters=1@172.18.1.176:9093,2@172.18.1.217:9093,3@172.18.1.150:9093

############################# Socket Server Settings #############################

# The address the socket server listens on.
# Combined nodes (i.e. those with `process.roles=broker,controller`) must list the controller listener here at a minimum.
# If the broker listener is not defined, the default listener will use a host name that is equal to the value of java.net.InetAddress.getCanonicalHostName(),
# with PLAINTEXT listener name, and port 9092.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://172.31.7.237:9092,CONTROLLER://172.18.1.217:9093

# Name of listener used for communication between brokers.
inter.broker.listener.name=PLAINTEXT

# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
#advertised.listeners=PLAINTEXT://172.18.1.217:9092


#完成后，生成整个集群有一个唯一的ID标志，使用uuid。可使用官方提供的 kafka-storage 工具生成
/opt/kafka/bin/kafka-storage.sh random-uuid

#用上述ID格式化存储路径
/opt/kafka/bin/kafka-storage.sh format -t clust_ID -c /opt/kafka/config/kraft/server.properties

#完成后以kraft模式启动服务
bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

#创建topic
bin/kafka-topics.sh --create --topic First_Kafka_Topic --partitions 1 --replication-factor 3 --bootstrap-server 172.31.7.237:9092

#查看
bin/kafka-topics.sh --list --bootstrap-server 172.31.7.237:9092

2）kafka启动脚本

#!/bin/bash
#kafka集群启动脚本
case $1 in
 "start"){
    for i in 172.18.1.176,172.18.1.217,@172.18.1.150
    do
       echo "--------启动 $i kafka with kraft-------"
       ssh $i "/home/kafka/bin/kafka-server-start.sh -daemon /home/kafka/config/kraft/server.properties"
    done
};;
"stop"){
    for i in 172.18.1.176,172.18.1.217,@172.18.1.150
    do
       echo "------停止 $i kafka--------"
       ssh $i "/home/kafka/bin/kafka-server-stop.sh"
    done
};;
esac