一、概要
Kafka作为一种高吞吐量的分布式发布订阅消息系统，在消息应用中广泛使用，尤其在需要实时数据处理和应用程序活动跟踪的场景，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模式的优势：
资料链接：官网kraft、kafka_jira
二、拓扑
2.1、早期拓扑
老controller选举原理： 只有一个broker可以当选Controller，当Controller在的broker宕机后，其他Broker可以竞争当选Controller。
常见几种副本状态：
当Kafka Topic刚刚创建时，该Topic的副本处于NonExistent状态，此时controller加载Zookeeper中该Topic每个分区的副本信息到内存中，同时将副本更新为New状态，之后controller选择分区的第一个副本作为leader副本并设置所有副本进入ISR，然后再Kafka中持久化该决定。
当确定了分区副本以及leader以后，controller会将这些信息发送给各个副本，同时将副本状态同步给所有的broker。上述操作完成以后，副本就会进入Online状态。
当开启了Topic删除操作，controller会停止所有副本，如果是follower副本将停止向leader副本fetch数据，如果是leader副本，controller会设置该分为的leade为NO_LEADER，之后副本进入Offline状态。紧接着，controller会将副本的状态变为ReplicaDeletionStarted状态表明开始进行Topic删除。controller会向所有的副本发出删除请求，请求他们删除本地的副本数据，当所有副本删除成功以后，便会进入ReplicaDeletionSuccessful状态。假设删除过程中有其中一个副本删除失败，则会进入ReplicaDeletionIneligible状态，等待controller的重试。同时处于ReplicaDeletionSuccessful状态会自动变为NonExistent状态，同时controller的上下文缓存会清除这些副本信息。
以下是分区状态：
当创建Topic时，controller负责创建分区对象，首先会短暂的将所有分区置为NonExistent状态，之后读取Zookeeper中的副本分配方案，然后令分区进入New状态。处于New状态的分区当选出Leader副本和ISR时，将会进入Online状态。
若用户发起了删除Topic操作，分区会进入NonExistent状态，并且还会开启分区下的副本删除操作。如果是关闭broker操作或者宕机，controller会判断broker是否是分区leader，如果是则需要开启新一轮的分区leader选举然后将分区状态改回Online状态。
2.2、kRaft controller拓扑
kafka kraft模式集群：必须是奇数节点，3节点最大容错1个，5节点容错2个
KRaft集群中，所有控制器代理都维护一个保持最新的内存元数据缓存，以便任何控制器都可以在需要时接管作为活动控制器（active controller），所有brokers都与控制器进行通信，其中活动控制器将处理与其他brokers通信对元数据的更改。KRaft 基于 Raft 共识协议，该协议作为 KIP-500 的一部分引入 Kafka，并在其他相关 KIP 中定义了更多细节。活动控制器是kafka kraft集群内部元数据主题的单个分区的leader，其他控制器是副本follower，brokers是副本观察者。因此，不是controlle将元数据更改广播给其他控制器或brokers，而是它们各自主动获取更改。这使得保持所有控制器和brokers同步非常有效，并且还缩短了broker和控制器的重启时间。
在 KRaft 模式下，集群元数据（反映所有控制器管理资源的当前状态）存储在名为__cluster_metadata. KRaft 使用这个主题在控制器和代理节点之间同步集群状态更改。KRaft 模式下，Kafka 集群可以以专用或共享模式运行。在专用模式下，一些节点将其process.roles配置设置为controller，而其余节点将其设置为broker。对于共享模式，这些节点将process.roles设置为controller, broker，即些节点将执行双重任务。该模式中那个称为“controller”的特殊节点负责管理集群中代理的注册，它一般就是controller.quorum.voters中的第一个节点。Broker 活跃度有两个条件：
Kradt模式下，the controllers store the cluster metadata in the directory specified in metadata.log.dir/the first log directory。特别注意，新增controller节点时需等待现有的controller提交完成所有数据后：The new controller node should not be formatted and started until the majority of the controllers have all of the committed data. 可通过如下命令来检查确认：kafka-metadata-quorum.sh --bootstrap-server broker_host:port describe --replication，如果显示Lag的值为0就最好，至少保证Lag值对于controllers足够小，或如果leader’s end offset is not increasing, you can wait until the lag is 0 for a majority；如果不是0，要检查下LastFetchTimestamp 和 LastCaughtUpTimestamp这2个值，所有controller的这2个值要尽量接近，即读到的和消费掉的要一致；之后才能执行：bin/kafka-storage.sh format --cluster-id uuid --config server_properties对新controller的元数据存储路径进行格式化；注意，混合模式下，存储格式化会报错：Log directory … is already formatted，这中情况仅出现在混合模式下且原controller 日志路径丢失或损坏的情况下，可执行：bin/kafka-storage.sh format --cluster-id uuid --ignore-formatted --config server_properties ，其他情况不建议使用。控制器用来接收来自其他控制器和borkers的请求。因此，即使服务器没有启用controller角色（即它只是一个broker），它仍然必须定义控制器侦听器以及配置它所需的任何安全属性。配置参考示例：
process.roles=broker
listeners=BROKER://localhost:9092 #broker does not expose the controller listener itself
inter.broker.listener.name=BROKER #区别于下面的controller.listener.names
controller.quorum.voters=0@localhost:9093
controller.listener.names=CONTROLLER #仅用于controller
listener.security.protocol.map=BROKER:SASL_SSL,CONTROLLER:SASL_SSL
#混合模式
process.roles=broker,controller
listeners=BROKER://localhost:9092,CONTROLLER://localhost:9093 #listeners独立配置用户kafka 客户端访问
inter.broker.listener.name=BROKER #配合上面的与kafka client交互，隔离开controller
controller.quorum.voters=0@localhost:9093
controller.listener.names=CONTROLLER #The controller will accept requests on all listeners defined by controller.listener.names，多个controller时，the first one in the list will be used for outbound requests.
listener.security.protocol.map=BROKER:SASL_SSL,CONTROLLER:SASL_SSL
更多参见 KRaft Principal Forwarding。注意：kraft模式下的混合模式多用于开发环境，生产环境不建议使用。混合模式下一个明显的问题就是， controller will be less isolated from the rest of the system，即无法与broker隔离，一个典型场景就是无法单独对controller进行弹性扩展或回滚升级；KRaft模式下，特定的Kafka服务器将被选作为controller；所有被定义候选的controller都会参与元数据的竞选，对已active的controller，其他的controller都充当hot standby 角色；因此，我们常用做法配置process.role 为 broker/controller而不是两者都具备（混合模式）；推荐，一个Kafka集群应该使用3个controller就可，超过3个也不推荐。Kafka控制器将集群的所有元数据存储在内存和磁盘上，官方建议知识给内存和磁盘分出各5G空间用来存储这些metadata日志；另外，kraft也有一些限制问题，比如以下功能未在KRaft模式中完全实现：
KRaft元数据复制过程：
集群元数据存储在 Kafka 主题中，而活动控制器是元数据主题的单个分区的领导者，它将接收所有数据并写入。其他控制器作为follower，将主动获取这些更改。对比传统的副本复制，当需要选举新leader时，是通过仲裁完成的，而不是同步副本集。因此，元数据复制不涉及 ISR。另一个区别是元数据记录在写入每个节点的本地日志时会立即刷新到磁盘。
Kraft 模式 以及 安全认证配置：
三、部署配置
wget https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
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 server acts as both a broker and a controller
# The node id associated with this instance's roles
node.id=2
# The connect string for the controller quorum 集群选举控制器配置，默认走 PLAINTEXT协议除非显示定义其他协议
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. it is used exclusively仅用于 for requests between brokers
inter.broker.listener.name=PLAINTEXT #注意与controller.listener.names不要冲突
# 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
或
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
#用上述ID格式化存储路径
/opt/kafka/bin/kafka-storage.sh format -t clust_ID -c /opt/kafka/config/kraft/server.properties
或
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c 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
3）kafka kRaft SASL/PLAIN 安全认证配置 ：
新建config 文件 config/kafka_server_jaas.conf，如下所示：
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
serviceName="kafka"
username="admin"
password="admin"
user_admin="admin";
};
复制一份kafka-server-start.sh ，修改名称 kafka-server-start-sasl.sh启动脚本修改名称，引入加密文件：
……
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/kafka_2.13-3.4.0/config/kafka_server_jaas.conf"
fi
……
复制一份 server_sasl.properties 修改名称为server_sasl.properties 与非安全认证分开，编辑：
node.id=1
controller.quorum.voters=1@kraft1:9093
listeners=SASL_PLAINTEXT://172.31.7.237:9092,CONTROLLER:///172.18.1.217:9093
inter.broker.listener.name=SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://172.31.7.237:9092
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
完成后启动kafka：sh ./bin/kafka-server-start-sasl.sh -daemon ./config/kraft/server_sasl.properties
四、附录：知识回顾
4.1、常用Message Queue对比
1）RabbitMQ
RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。
2）Redis
Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。
3）ZeroMQ
ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。
4）ActiveMQ
ActiveMQ是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。
5）Kafka/Jafka
Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。kafka3.0轻量级的单进程部署不仅可以替代ActiveMQ、RabbitMQ等传统的消息队列，同时也适合于边缘场景和使用轻量级硬件的场景。数据表明，一个可以管理 两百万分区的集群中，kafka3.0新版本中Quorum Controller的迁移过程可以 从几分钟缩小至30秒，突破了限制Kafka集群范围的元数据管理这个主要瓶颈，关机和重启 十多倍速于kafka2.8，性能完全实现了强力碾压！
4.2、Kafka的原理
Kafka的一种新的复制协议，它是在Kafka2.4版本中引入的。Kraft协议的目标是提高Kafka的可靠性和可维护性，满足分布式流处理需求；它主要包括以下几个方面：
4.3、DoctorKafka 管理工具
它是源于Pinterest产品的一个项目，为了扩展Kafka 服务的运维规模，Pinterest 构建了DoctorKafka，用于管理 Kafka 集群自愈和工作负载均衡；。DoctorKafka 能够探测到Kafka broker 的故障并自动将故障broker 的负载转移给健康的broker。现在，Pinterest 已经在 GitHub 上将该项目开源。DoctorKafka 由三部分组成,架构如下：
DoctorKafka 服务启动之后，它会首先读取 broker 最近 24 到 48 小时的状态，基于此，DoctorKafka 会推断每个副本工作负载所需的资源。因为 Kafka 工作负载主要是网络密集型的，DoctorKafka 主要关注副本的网络带宽使用情况。DoctorKafka 在启动之后，会阶段性地检查每个集群的状态。当探测到 broker 出现故障时，它会将故障 broker 的工作负载转移给有足够带宽的 broker。如果在集群中没有足够的资源进行重分配的话，它会发出告警。与之类似，当 DoctorKafka 进行工作负载平衡时，它会识别出网络流量超出配置的 broker，并将工作负载转移给流量更少的 broker，或者是执行更优的领导者选举（leader election）方案来转移流量。
DoctorKafka 已经在 Pinterest 运行了数月之久，并帮助其运维人员管理着 1000 个以上的集群。更多参见 项目地址。
4.4、Kafka WebUI之Kowl
Kowl ( Kafka Owl)： Kafka WebUI for exploring messages, consumers. configurations and more with a focus on a good UI。更多参看：项目地址。
除上之外，还有9种常见的kafka UI工具，除此之外还有LogiKM 和 kafka-console-ui ，因Kafka-UI 功能齐全且免费，推荐。
一般该程序通过docker容器运行，Kowl 的最大优势在于其出色的用户界面。它方便、用户友好且易于使用；启动后界面如下：
Kowl 提供消息浏览、实时跟踪以及对 Protobuf、Avro 和 Amazon MSK IAM 的支持,但登录系统(Google、GitHub、Okta)和具有组同步的 RBAC 权限仅适用于付费的 Kowl Business 计划。Kowl 还缺少多集群管理、动态主题配置、分区增加、副本更改、Kafka Connect 管理、模式注册表、KSQL 集成、Kafka Streams 拓扑、只读模式以及 JMX 指标的可视化和图表等功能。
