[大数据]玩转 Kafka Raft 模式 - 入门宝典

Apache Kafka 将会在3.0 的版本里去掉 Apache Zookeeper 的依赖独立运行。3.0的的Kafka 从架构，代码层面相对于1.*，2.*版本 发生了很大的变化。变化主要集中在Kafka Raft，Contrller Quorum， Metadata，Snapshot等模块的改进。本文是Kafka Raft技术系列中的第一篇，我们先来初步体验一下KRaft的入门使用。

作者介绍：

许文强：
腾讯高级工程师
腾讯云CKafka研发负责人，Apache Kafka Contributor
拥有多年分布式系统研发经验，主要负责腾讯云CKafka定制化开发及优化工作。专注于Kafka在公有云多租户和大规模集群场景下的性能分析和优化。

KRaft 简介

Apache Kafka 不依赖 Apache Zookeeper的版本，被社区称之为Kafka Raft 元数据模式，简称KRaft(craft)模式。该模式在2.8当中已经发布了体验版本。可以初步体验KRaft的运行效果，但是还不建议在生产环境中使用。未来3.0会出一个稳定的release版本。

KRaft运行模式的Kafka集群 ，不会将元数据存储在 Apache ZooKeeper中。 即部署新集群的时候，无需部署ZooKeeper集群 ，因为 Kafka 将元数据存储在 Controller 节点的 KRaft Quorum中 。 KRaft可以带来很多好处，比如可以支持更多的分区，更快速的切换Controller，也可以避免Controller缓存的元数据和Zookeeper存储的数据不一致带来的一系列问题。

KRaft 架构

首先来看一下KRaft在系统架构层面和之前的版本有什么区别。KRaft模式提出了去 Zookeeper后的Kafka整体架构如下，下图是前后的架构图对比：

在当前架构中，Kafka集群包含多个broker节点和一个ZooKeeper 集群。 我们在这张图中描绘了一个典型的集群结构： 4个broker节点和3个ZooKeeper节点。 Kafka 集群的Controller (橙色)在被选中后，会从 ZooKeeper 中加载它的状态。 Controller 指向其他 Broker 节点的箭头表示 Controller 在通知其他Broker发生了变更，如Leaderanddisr和Updatemetdata请求。

在新的架构中，三个 Controller 节点替代三个ZooKeeper节点。 控制器节点和 Broker 节点运行在不同的进程中。 Controller 节点中会选择一个成为Leader(橙色)。 新的架构中，控制器不会向 Broker 推送更新，而是 Broker 从这个 Controller Leader 拉取元数据的更新信息。

需要特别注意的是，尽管 Controller 进程在逻辑上与 Broker 进程是分离的，但它们不需要在物理上分离。 即在某些情况下，部分或所有 Controller 进程和 Broker 进程是可以是同一个进程，即一个broker节点即是Broker也是Controller。 另外在同一个节点上可以运行两个进程，一个是Controller进程，一个是Broker进程，这相当于在较小的集群中，ZooKeeper进程可以像Kafka Broker一样部署在相同的节点上。

接下来，我们来看一下如何运行我们的第一个KRaft集群：

部署和配置

在当前架构中，新增了如下三个参数：

# 标识该节点所承担的角色，在KRaft模式下需要设置这个值
process.roles=broker,controller

# 节点的ID，和节点所承担的角色相关联
node.id=1

# controller quorum 连接的集群地址字符串
controller.quorum.voters=1@localhost:9093

Controller 服务器

在KRaft模式下，只有一小部分特别指定的服务器可以作为控制器，在Server.properties的Process.roles 参数里面配置。不像基于ZooKeeper的模式，任何服务器都可以成为控制器。这带来了一个非常优秀的好处，即如果我们认为Controller节点的负载会比其他只当做Broker节点高，那么配置为Controller节点就可以使用高配的机器。这就解决了在1.0, 2.0架构中，Controller节点会比其他节点负载高，却无法控制哪些节点能成为Controller节点的问题。

被选中的 Controller 节点将参与元数据集群的选举。对于当前的Controller节点，每个控制器服务器要么是Active的，要么是Standby的。

用户通常会选择3或5台(奇数台)服务器成为Controller节点，3和5的个数问题和Raft的原理一样，少数服从多数。这取决于成本和系统在不影响可用性的情况下应该承受的并发故障数量等因素。对Raft协议熟悉的同学们应该可以理解这块的逻辑，不熟悉的同学也可以去了解一下Raft。

就像使用ZooKeeper一样，为了保持可用性，你必须让大部分Controller保持active状态。如果你有3个控制器，你可以容忍1个故障;在5个控制器中，您可以容忍2个故障。

Process.Roles

每个Kafka服务器现在都有一个新的配置项，叫做Process.Roles, 这个参数可以有以下值:

• 如果Process.Roles = Broker, 服务器在KRaft模式中充当 Broker。
• 如果Process.Roles = Controller, 服务器在KRaft模式下充当 Controller。
• 如果Process.Roles = Broker,Controller，服务器在KRaft模式中同时充当 Broker 和Controller。
• 如果process.roles 没有设置。那么集群就假定是运行在ZooKeeper模式下。

如前所述，目前不能在不重新格式化目录的情况下在ZooKeeper模式和KRaft模式之间来回转换。 同时充当Broker和Controller的节点称为“组合”节点。

对于简单的场景，组合节点更容易运行和部署，可以避免多进程运行时，JVM带来的相关的固定内存开销。 关键的缺点是，控制器将较少地与系统的其余部分隔离。 例如，如果代理上的活动导致内存不足，则服务器的控制器部分不会与该OOM条件隔离。

Quorum Voters

系统中的所有节点都必须设置 controller.quorum.voters 配置。这个配置标识有哪些节点是 Quorum 的投票者节点。所有想成为控制器的节点都需要包含在这个配置里面。这类似于在使用ZooKeeper时，使用ZooKeeper.connect配置时必须包含所有的ZooKeeper服务器。

然而，与ZooKeeper配置不同的是，controller.quorum.voters 配置需要包含每个节点的id。格式为: id1@host1:port1,id2@host2:port2。

因此，如果你有10个Broker和 3个控制器，分别命名为Controller1、Controller2、Controller3，你可能在Controller1上有以下配置:

process.roles = controller
node.id = 1
listeners=CONTROLLER://controller1.example.com:9093
controller.quorum.voters=1@controller1.example.com:
9093,2@controller2.example.com:9093,3@controller3.example.com:
9093

每个Broker和每个Controller 都必须设置 controller.quorum.voters。需要注意的是，controller.quorum.voters 配置中提供的节点ID必须与提供给服务器的节点ID匹配。

比如在Controller1上，node.Id必须设置为1，以此类推。注意，控制器id不强制要求你从0或1开始。然而，分配节点ID的最简单和最不容易混淆的方法是给每个服务器一个数字ID，然后从0开始。

运行KRaft集群

运行KRaft集群，主要分为三步：

• 用kafka-storage.sh 生成一个唯一的集群ID
• 用kafka-storage.sh 格式化存储数据的目录
• 用bin/kafka-server-start.sh 启动Kafka Server

在1.0和2.0的版本里面，集群ID是自动生成的，存储数据目录是自动生成的。那为什么在3.0会这样做呢？

社区的的思考是这样子的，即自动格式化有时候会掩盖一些异常，比如，在Unix中，如果一个数据目录不能被挂载，它可能显示为空白，在这种情况下，自动格式化将是将会带来一些问题。这个特性对于 Controller 服务器维护元数据日志特别重要，因为如果三个 Controller 节点中有两个能够从空白日志开始，那么可能会在日志中没有任何内容的情况下，选出一个Leader，这会导致所有的元数据丢失(KRaft 仲裁后发生截断)。一旦发生这个问题，将会是不可逆的故障。

生成集群ID

首先是使用bin目录下的kafka-storage.sh工具为你的新集群生成一个唯一的ID，

$ ./bin/kafka-storage.sh random-uuid 
xtzWWN4bTjitpL3kfd9s5g

格式化存储目录

接着是格式化存储目录。如果是单节点模式运行，你需要在机器上执行如下命令。如果是多个节点，则应该在每个节点上都分别运行format命令，以便格式化每台机器上的。请确保为每个集群使用相同的集群ID。

$ ./bin/kafka-storage.sh format -t <uuid> -c 
./config/kraft/server.properties Formatting /tmp/kraft-combined-
logs

启动Kafka Server

最后，可以在每个节点上启动Kafka服务器了。

$ ./bin/kafka-server-start.sh ./config/kraft/server.properties 

就像基于ZooKeeper的 集群 一样，可以连接到端口9092(或配置的任何端口)来执行
理操作 ，如创建删除Topic，也可以用原先命令进行生产消费 。

如创建Topic：

$ ./bin/kafka-topics.sh --create --topic foo --partitions 1 --
replication-factor 1 --bootstrap-server localhost:9092 Created
 topic foo.

生产消费信息：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic foo
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092-
from-beginning --topic  foo  --group foo_group

实用工具

使用过程中，如果遇到问题，可能需要查看元数据日志。在KRaft中，有两个命令行工具需要特别关注下。kafka-dump-log.sh和kakfa-metadata-shell.log。

Kafka-dump-log.sh

KRaft模式下 ，原先保存在Zookeeper上的数据，全部转移到了一个内部的Topic：@metadata上了。比如Broker信息，Topic信息等等。所以我们需要有一个工具查看当前的数据内容。

Kafka-dump-log.sh是一个之前就有的工具，用来查看Topic的的文件内容。这工具加了一个参数–cluster-metadata-decoder用来，查看元数据日志，如下所示:

$ ./bin/kafka-dump-log.sh  --cluster-metadata-decoder --skip-
record-metadata --files /tmp/kraft-combined-
logs/\@metadata-0/*.log
Dumping /tmp/kraft-combined-
logs/@metadata-0/00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 
lastSequence: -1 producerId: -1 producerEpoch: -1 
partitionLeaderEpoch: 1 isTransactional: false isControl: true
position: 0 CreateTime: 1614382631640 size: 89 magic: 2 
compresscodec: NONE crc: 1438115474 isvalid: true

Kafka-metadata-shell.sh

平时我们用zk的时候，习惯了用zk命令行查看数据，简单快捷。bin目录下自带了kafka-metadata-shell.sh工具，可以允许你像zk一样方便的查看数据。

$ ./bin/kafka-metadata-shell.sh  --snapshot /tmp/kraft-combined-
logs/\@metadata-0/00000000000000000000.log
>> ls /
brokers  local  metadataQuorum  topicIds  topics
>> ls /topics
foo
>> cat /topics/foo/0/data
{
  "partitionId" : 0,
  "topicId" : "5zoAlv-xEh9xRANKXt1Lbg",
  "replicas" : [ 1 ],
  "isr" : [ 1 ],
  "removingReplicas" : null,
  "addingReplicas" : null,
  "leader" : 1,
  "leaderEpoch" : 0,
  "partitionEpoch" : 0
}
>> exit

总结

Kafka 经常被认为是一个重量级的基础设施，管理Apache Zookeeper的复杂性就是这种看法存在的重要原因。这通常会导致项目在开始时选择更轻量级的消息队列，比如ActiveMQ或Rabbitmq这样的传统消息队列，然后在规模变大时迁移到Kafka。

现在已经不是这样了。KRaft模式提供了一种很棒的、轻量级的方式来开始使用Kafka，或者可以使用它作为ActiveMQ或RabbitMQ等消息队列的替代方案。轻量级的单进程部署也更适合于边缘场景和那些使用轻量级硬件的场景。

在后续的文章中，我们会详细展开KRaft、Controler Quorum，Kafka Raft Snapshot相关的原理解析和代码讲解。让大家可以提前一步熟悉KRaft 3.0。