只需三步，使用 KRaft 建立多节点 Kafka 集群

Apache Kafka是一个用 Java 编写的开源分布式事件和流处理平台，用于处理要求苛刻的实时数据馈送。它本质上是可扩展的，具有高吞吐量和高可用性。其设计也具有容错性，每个集群可支持数百个节点。

在本教程中，你将创建一个 Kafka 集群，使用 KRaft共识协议的 Kafka 集群。你将学习如何配置节点成为集群的一部分，并观察主题分区是如何分配给不同节点的。你还将学习如何将主题分配给集群中的特定代理。

让我们开始吧！

准备工作

为了你能顺利跟上本教程的步骤，你需要这些：

三个可用的 Droplets，至少需要 4GB 内存以及 2 个 CPU。如果是 Ubuntu 服务器，请按照 Ubuntu 初始服务器设置获取设置说明，或参照教程在 DigitalOcean Droplet 云主机上创建 Ubuntu 服务器。
一个完全注册的域名，带有指向三个 Droplets的三个子域。本教程将把它们分别称为 kafkaX.your_domain。你可以在 Namecheap 购买域名，在 Freenom 上免费获取一个，或者也可以使用你选择的域名注册商。
在 Droplets 上安装和配置 Apache Kafka。有关安装说明，你可以按照 Kafka 简介教程配置。你只需完成第一步 和第二步。

第一步：配置 Kafka 节点

在这一步中，你将配置作为准备工作

一部分创建的三个 Kafka 服务器，使其成为同一个 KRaft 集群的一部分。有了 KRaft，节点本身就可以组织和执行管理任务，而无需依赖 Apache ZooKeeper。

配置第一个节点

你将先配置第一个节点。首先，运行以下命令停止第一个 Droplet 上的服务：

sudo systemctl stop kafka

接下来，以用户 kafka 的身份，导航到 Kafka 所在的目录，并通过运行打开其配置文件进行编辑：

vi /config/kraft/server.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=1# The connect string for the controller quorum
controller.quorum.voters=1@localhost:9092
...

这三个参数将 Kafka 节点同时配置为代理和控制器，这意味着它将接收和消费数据（代理）并执行管理任务（控制器）。这种分离在大型部署中非常有用，可以将控制器分开，以提高效率和冗余。

node.id 指定了节点在群集中的 ID。这是第一个节点，所以它的 ID 是 1，因此第二个和第三个节点的 ID 将分别为 2 和 3。

controller.quorum.voters 将节点 ID 映射到它们各自的地址和端口，以便通信。这里将指定所有集群节点的地址，以便每个节点都能知道其他节点。修改该行如下(config/kraft/server.properties)：

...
controller.quorum.voters=1@kafka1.your_domain:9093,2@kafka2.your_domain:9093,3@kafka3.your_domain:9093
...

在此，你将列出集群中的所有三个节点及其各自的 ID。记住将 your_domain 替换为你在准备工作中设置的域名地址。

接下来，在文件中找到以下几行：

...
listeners=PLAINTEXT://:9092,CONTROLLER://: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://localhost:9092
...

listeners 定义了 Kafka 节点监听的地址，而 advertised.listeners 则指定了将传递给客户端以连接到节点的地址。这让你可以为客户端指定它应该使用的实际地址的子集。

修改以下代码行，用实际域名替换其中的 your_domain：

...
listeners=PLAINTEXT://kafka1.your_domain:9092,CONTROLLER://kafka1.your_domain: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://kafka1.your_domain:9092
...

由于该节点将位于一个集群中，因此你已明确将地址指向了运行该节点的 Droplet。

然后，找到 num.partitions设置：

...
# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1
...

正如注释所述，这配里置了每个新主题的默认分区数量。由于你有三个节点，因此将其设置为 2 的倍数：

...
num.partitions=6
...

这里的值为 6 可确保每个节点默认拥有两个主题分区。

下一步是配置内部主题的复制因子，它将保留消费者偏移和事务状态。找到以下几行：

...
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
...

将它们设置为以下值：

...
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
...

在这里，你需要指定至少两个节点的内部元数据必须同步。完成后，保存并关闭文件。

设置默认分区号后，必须重新初始化日志存储。首先，运行以下命令删除现有日志文件：

rm -rf /home/kafka/kafka-logs/*

然后，生成一个新的群集 ID 并将其存储为环境变量：

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

在终端显示：

echo $KAFKA_CLUSTER_ID

输出将是集群 ID：


Mjj4bch9Q3-B0TEXv8_zPg

请注意该值；配置第二个和第三个节点时会用到它。

最后，运行以下命令生成日志存储：

./bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

输出结果与此类似：

...
Formatting /home/kafka/kafka-logs with metadata.version 3.7-IV4.

配置第二和第三节点

其他节点的配置与第一个节点的配置非常相似。注意也要更新 node.id：

...
node.id=node_number
...

第二个和第三个节点的值分别为2 和 3，并为 listeners 和 advertised.listeners 设置地址。

重新生成日志存储时，重复使用第一个节点的群集 ID：

KAFKA_CLUSTER_ID="your_cluster_id"

完成后，在所有三个节点上运行 Kafka 服务：

sudo systemctl start kafka

在这一步中，你已将三个 Kafka 节点配置为 KRaft 集群的一部分。你将创建一个主题，并在集群上生产和消费消息。

第二步：连接至群集

在这一步中，你将使用 Kafka 捆绑的 shell 脚本连接到 Kafka 集群。你还将创建一个主题，并尝试从集群中生产和消费数据。然后，你将使其中一个节点宕机，并观察 Kafka 如何减轻损失。

Kafka 提供了kafka-metadata-quorum.sh 脚本，用于显示有关集群及其成员的信息。运行以下命令执行该脚本：

./bin/kafka-metadata-quorum.sh --bootstrap-controller kafka1.your_domain:9093 describe --status

你通过 9093 端口连接到其中一个节点，这是控制器的端点（但不是代理的端点）。切记将 kafka1.your_domain 替换为指向其中一个 Kafka 节点的域。

输出结果为：

ClusterId:              G3TeIZoeTSCvG2YOWvPE2w
LeaderId:               3
LeaderEpoch:            2
HighWatermark:          383
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   55
CurrentVoters:          [1,2,3]
CurrentObservers:       []

脚本列出了集群状态的基本信息。在显示的输出中，你可以看到节点 3 被选为领导者，所有三个节点（[1,2,3]）都在投票池中，并同意这一决定。

运行并创建一个名为 first-topic 的主题：

./bin/kafka-topics.sh --create --topic first-topic --bootstrap-server kafka1.your_domain:9092 --replication-factor 2

输出结果将是：

Created topic first-topic.

然后，运行 kafka-topics.sh 脚本，查看分区在节点上的排列情况：

./bin/kafka-topics.sh --describe --bootstrap-server kafka1.your_domain:9092 --topic first-topic

将复制因子设置为 2 可确保主题至少在两个节点上可用。

输出结果与此类似：

Topic: first-topic      TopicId: 4kVImoFNTQeyk3r2zQbdvw PartitionCount: 6       ReplicationFactor: 2    Configs: segment.bytes=1073741824Topic: first-topic      Partition: 0    Leader: 3       Replicas: 3,1   Isr: 3,1Topic: first-topic      Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2Topic: first-topic      Partition: 2    Leader: 2       Replicas: 2,3   Isr: 2,3Topic: first-topic      Partition: 3    Leader: 1       Replicas: 1,3   Isr: 1,3Topic: first-topic      Partition: 4    Leader: 3       Replicas: 3,2   Isr: 3,2Topic: first-topic      Partition: 5    Leader: 2       Replicas: 2,1   Isr: 2,1

可以看到，每个分区都有一个领导者、两个副本和两个同步副本集（ISR）。分区领导者是向客户端提供分区数据的代理节点，而副本节点只保存副本。默认情况下，如果一个副本节点在过去的十秒钟内赶上了领导者，那么它就被认为是 ISR。这个时间间隔可根据每个主题进行配置。

创建主题后，就可以使用 kafka-console-producer.sh 脚本生成消息了。运行以下命令启动生产者：

./bin/kafka-console-producer.sh --topic first-topic --bootstrap-server kafka1.your_domain:9092

你将看到一个空提示：

生产者正等着你输入文字信息。输入 test 并按 ENTER 键。提示如下

>Hello World!
>

生产者现在正在等待下一条消息，这意味着上一条消息已成功传送到 Kafka。你可以输入任意数量的信息进行测试。要退出生产者，请按 CTRL+C。

你需要一个消耗者来回读主题中的消息。Kafka 提供了一个简单的消耗者，名为 kafka-console-consumer.sh。通过执行以下命令运行:

./bin/kafka-console-consumer.sh --topic first-topic --from-beginning --bootstrap-server kafka1.your_domain:9092

你将看到从主题中读取的信息：

Hello World!
...

模拟节点故障

在第三个 Kafka 节点上，运行以下命令停止服务：

sudo systemctl stop kafka

然后，运行描述主题：

./bin/kafka-topics.sh --describe --bootstrap-server kafka1.your_domain:9092 --topic first-topic

输出结果与此类似：

Topic: first-topic      TopicId: 4kVImoFNTQeyk3r2zQbdvw PartitionCount: 6       ReplicationFactor: 2    Configs: segment.bytes=1073741824Topic: first-topic      Partition: 0    Leader: 1       Replicas: 3,1   Isr: 1Topic: first-topic      Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2Topic: first-topic      Partition: 2    Leader: 2       Replicas: 2,3   Isr: 2Topic: first-topic      Partition: 3    Leader: 1       Replicas: 1,3   Isr: 1Topic: first-topic      Partition: 4    Leader: 2       Replicas: 3,2   Isr: 2Topic: first-topic      Partition: 5    Leader: 2       Replicas: 2,1   Isr: 2,1

尽管节点 3 已被列为副本，但由于它不可用，所以 ISR 集中缺少它。一旦重新加入群集，它将与其他节点同步，并尝试重新获得之前的位置。

再试着阅读一下 first-topic 中的信息：

./bin/kafka-console-consumer.sh --topic first-topic --from-beginning --bootstrap-server kafka1.your_domain:9092

你会发现它们可以像往常一样访问：

Hello World!
...

由于复制的存在，前两个节点会接管并为消费者提供服务。现在，你可以在第三台服务器上启动 Kafka：

sudo systemctl start kafka

在这一步中，你已经看到 Kafka 如何缓解集群中节点的不可用性。现在，你将学习如何从容地从集群中排除一个节点。

第三步：在节点之间迁移数据

在本步骤中，你将学习如何在 Kafka 集群的节点之间迁移主题。在向包含主题的现有集群添加节点时，Kafka 不会自动将任何分区转移到集群中，而你可能希望这样做。这对于删除节点也很有用，因为现有分区不会自动转移到剩余的节点上。

Kafka 提供了一个名为 kafka-reassign-partitions.sh 的脚本，它可以生成、执行和验证转换计划。你将使用它创建一个计划，将 first-topic 的分区移动到前两个节点。

首先，你需要定义要移动的主题。脚本接受一个包含主题定义的 JSON 文件，因此请创建并打开该文件进行编辑：

vi topics-to-move.json

添加以下几行 :

{"topics": [{"topic": "first-topic"}],"version": 1
}

在topic下，定义一个引用first-topic的对象。完成后，保存并关闭文件。

运行以下命令生成迁移计划，将 kafka1.your_domain 替换为指向其中一个 Kafka 节点的域：

./bin/kafka-reassign-partitions.sh --bootstrap-server kafka1.your_domain:9092 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate

在 --broker-list 中输入 "1,2"，表示目标broker的 ID。

输出结果与此类似：

Current partition replica assignment
{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[3,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[2,1],"log_dirs":["any","any"]}]}Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[1,2],"log_dirs":["any","any"]}]}

脚本总共生成了两个计划，分别描述了当前和建议的分区布局。如果以后需要恢复更改，我们会提供第一个计划。请注意第二个计划，它将保存在一个名为 migration-plan.json 的单独文件中。创建并打开该文件进行编辑：

vi migration-plan.json

添加第二个执行计划，在migration-plan.json 中输入：

{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[1,2],"log_dirs":["any","any"]}]}

保存并关闭文件。然后，运行以下命令执行该文件：

./bin/kafka-reassign-partitions.sh --bootstrap-server kafka1.your_domain:9092 --reassignment-json-file migration-plan.json --execute

输出结果将是

Current partition replica assignment{"version":1,"partitions":[{"topic":"first-topic","partition":0,"replicas":[3,1],"log_dirs":["any","any"]},{"topic":"first-topic","partition":1,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":2,"replicas":[2,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":3,"replicas":[1,3],"log_dirs":["any","any"]},{"topic":"first-topic","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{"topic":"first-topic","partition":5,"replicas":[2,1],"log_dirs":["any","any"]}]}Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for first-topic-0,first-topic-1,first-topic-2,first-topic-3,first-topic-4,first-topic-5

脚本指出迁移已经开始。要查看迁移的进度，请使用 --verify 代替：

./bin/kafka-reassign-partitions.sh --bootstrap-server kafka1.your_domain:9092 --reassignment-json-file migration-plan.json --verify

一段时间后，输出结果将与此相似：

Status of partition reassignment:
Reassignment of partition first-topic-0 is completed.
Reassignment of partition first-topic-1 is completed.
Reassignment of partition first-topic-2 is completed.
Reassignment of partition first-topic-3 is completed.
Reassignment of partition first-topic-4 is completed.
Reassignment of partition first-topic-5 is completed.Clearing broker-level throttles on brokers 1,2,3
Clearing topic-level throttles on topic first-topic

现在你可以描述 first-topic，以验证代理 3 上没有分区：

./bin/kafka-topics.sh --describe --bootstrap-server kafka1.your_domain:9092 --topic first-topic

输出结果如下

Topic: first-topic      TopicId: 4kVImoFNTQeyk3r2zQbdvw PartitionCount: 6       ReplicationFactor: 2    Configs: segment.bytes=1073741824Topic: first-topic      Partition: 0    Leader: 2       Replicas: 2,1   Isr: 1,2Topic: first-topic      Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2Topic: first-topic      Partition: 2    Leader: 2       Replicas: 2,1   Isr: 2,1Topic: first-topic      Partition: 3    Leader: 1       Replicas: 1,2   Isr: 1,2Topic: first-topic      Partition: 4    Leader: 2       Replicas: 2,1   Isr: 2,1Topic: first-topic      Partition: 5    Leader: 1       Replicas: 1,2   Isr: 2,1

只有broker 1 和 2 作为副本和 ISR 存在，这意味着迁移成功。

在本步骤中，你已制定了将第一个主题从代理 3 迁移到其余主题的迁移计划，并学会了如何验证迁移是否顺利进行。

结论

你现在有了一个 Kafka 集群，它由三个节点组成，使用 KRaft 协议进行通信。你还学会了如何检查集群和分区布局。你已经通过关闭一个节点并从一个主题读取数据来测试集群的冗余性。最后，你还学会了如何将主题重新分配给集群中的节点。

最后，欢迎了解 DigitalOcean 托管 Kafka 数据库服务，DigitalOcean 的 Kafka 支持横向扩展，并且支持用户对 Kafka 自定义指标进行后续的数据监控与分析。如需了解更多 DigitalOcean 云服务的数据产品或其它技术产品，可访问 DigitalOcean 中国区独家战略合作伙伴卓普云的官网。