前述

在本章节中，我们将实践 Kafka 的基本操作。首先，会创建一个新的主题（Topic）并执行消息的生产与消费。随后，将模拟一次节点故障，通过主动关闭集群中的一个 Broker，来检验 Kafka 集群的容错能力和服务的连续性。

主题管理

创建主题

创建主题（Topic）one-topic，并设置副本数为 2，检测主题、分区信息。

# 创建主题
[root@kafka-node-001 kafka]# /opt/kafka/bin/kafka-topics.sh --create --topic one-topic --bootstrap-server kafka-node-001:9092 --replication-factor 2
Created topic one-topic.

# 查看分区信息
[root@kafka-node-001 kafka]# /opt/kafka/bin/kafka-topics.sh --describe --bootstrap-server kafka-node-001:9092 --topic one-topic
Topic: one-topic      TopicId: gFMGRZuqRv6y4qXgvdkaFw PartitionCount: 6       ReplicationFactor: 2    Configs: min.insync.replicas=1,segment.bytes=1073741824
        Topic: one-topic      Partition: 0    Leader: 3       Replicas: 3,1   Isr: 3,1        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 2    Leader: 2       Replicas: 2,3   Isr: 2,3        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 3    Leader: 3       Replicas: 3,2   Isr: 3,2        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 4    Leader: 2       Replicas: 2,1   Isr: 2,1        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 5    Leader: 1       Replicas: 1,3   Isr: 1,3        Elr:    LastKnownElr: 

kafka-topics.sh --describe 是非常有用的命令，可以让您了解一个 Topic 在 Kafka 集群中的健康状况和分布情况。这个命令的输出分为两个部分：

1. 主题摘要信息：第一行是对整个 Topic 的总结。
2. 分区详细信息：后续的每一行都详细描述了该 Topic 的每个分区（Partition）。

主题摘要信息

Topic: one-topic  TopicId: gFMGRZuqRv6y4qXgvdkaFw  PartitionCount: 6  ReplicationFactor: 2  Configs: min.insync.replicas=1,segment.bytes=1073741824

字段分解：

• Topic: one-topic: 主题的名称 one-topic。
• TopicId: gFMGRZuqRv6y4qXgvdkaFw: 主题的唯一 ID。即使删除这个 Topic，然后再创建一个同名的，它的 TopicId 也会不同。这是 Kafka 内部用来唯一标识主题的。
• PartitionCount: 6: 分区数量。这个主题被分成了 6 个分区（编号从 0 到 5）。分区是 Kafka 实现并行处理和提高吞吐量的基本单位。
• ReplicationFactor: 2: 副本因子/复制因子。这表示每个分区都有 2 个副本（1 个 Leader 和 1 个 Follower）。副本是为了实现数据的高可用性和容错能力。如果一个副本所在的 broker 宕机，其他副本可以继续提供服务。
• Configs: min.insync.replicas=1,segment.bytes=1073741824: 为该主题指定的特殊配置。
  • min.insync.replicas=1: 这是非常重要的一个持久性保证参数。它规定了当生产者设置 acks=all 时，一条消息需要被至少多少个“同步副本”（In-Sync Replicas）确认后，才算写入成功。这里设置为 1，意味着只要 Leader 副本写入成功，生产者就会收到确认。注意： 这个配置的容错性较低，如果 Leader 刚写入就宕机，而 Follower 还没来得及同步，可能会导致消息丢失。对于高可靠性场景，通常会设置为 2（即至少需要 Leader + 一个 Follower 都确认）。
  • segment.bytes=1073741824: 这是日志段文件的大小，1073741824 字节等于 1 GB。当分区的日志文件达到这个大小时，Kafka 会切分出一个新的日志段文件。这是一个性能调优参数。

分区详细信息

接下来以分区 0 为例，详细解释每一列的含义。

Topic: one-topic  Partition: 0  Leader: 3  Replicas: 3,1  Isr: 3,1  Elr:  LastKnownElr: 

• Partition: 0: 分区的编号是 0 号。
• Leader: 3: 当前为该分区提供读写服务的 Leader 副本 所在的 Broker ID 是 Broker 3。所有的生产者请求和消费者请求都首先发往这个 Leader 副本。
• Replicas: 3,1: 该分区所有副本所在的 Broker ID 列表。这里表示分区 0 的数据在 Broker 3 和 Broker 1 上各有一份。Leader 总是这个列表的成员之一。
• Isr: 3,1: In-Sync Replicas（同步副本集）的 Broker ID 列表。这是 Replicas 列表的一个子集，表示那些与 Leader 副本保持“完全同步”或“基本同步”的副本。只有 ISR 列表中的副本才有资格在当前 Leader 宕机时被选举为新的 Leader。
  • 健康状态的关键指标：在这里，Isr 列表（3,1）和 Replicas 列表（3,1）完全一样，这表明 Broker 1 上的 Follower 副本与 Broker 3 上的 Leader 副本保持着良好的同步状态。这是一个非常健康的信号。如果 Isr 列表比 Replicas 列表少成员（例如 Isr 只有 3），则说明 Broker 1 出现了延迟或无响应，此时该分区处于 “欠复制(Under-replicated)” 状态，存在风险。
• Elr: 和 LastKnownElr: : 这两个字段与 分层存储（Tiered Storage） 相关，分别代表 “合格的 Leader 副本（Eligible Leader Replicas）” 和 “最后已知的合格 Leader 副本”。它们用于在分层存储环境中，优化 Leader 选举，优先选择那些本地有完整数据的副本。这里为空，表示你的 Kafka 集群没有为这个 Topic 启用分层存储功能，或者版本较低。

小结

1. 集群规模: Kafka 集群至少有 3 台 Broker（因为我们看到了 Broker ID 1, 2, 3）。
2. 负载均衡: 6 个分区的 Leader 副本被相对均匀地分布在了 Broker 1, 2, 3 上，这是一个良好的分布，有助于分摊读写压力。
3. 健康状况: 非常健康，对于所有的 6 个分区，Isr 列表都和 Replicas 列表完全一致。这说明所有副本都处于同步状态，集群具有预期的容错能力。
4. 数据可靠性: min.insync.replicas=1 的配置意味着在生产者端追求的是较低的延迟而非最高的可靠性。在 acks=all 的情况下，只要 Leader 写入成功就返回，不等待 Follower 同步。如果对数据丢失零容忍，应考虑将此值提高到 2。

最小同步副本数设置

沿用上述的建议，可以将最小同步副本数设置为 2。

# 修改现有主题的最小同步副本数
[root@kafka-node-001 kafka]# /opt/kafka/bin/kafka-topics.sh --alter --bootstrap-server kafka-node-001:9092 --topic one-topic --config min.insync.replicas=2

# 在创建主题的时候，直接设置最小同步副本数
[root@kafka-node-001 kafka]#。/opt/kafka/bin/kafka-topics.sh --create --bootstrap-server kafka-node-001:9092 \
--topic highly-available-topic \ # 主题名称
--partitions 3 \ # 分区数量，我们已经在 server.properties 中设置为 6 
--replication-factor 3 \ # 副本数量
--config min.insync.replicas=2 # 最小同步副本数

消息收发测试

通过 kafka-node1 节点发送消息，然后在 kafka-node2 打开接收器，看是否可以正常进行消息的收发。

# 使用消息提供者脚本，指定主题名称，节点名称，回车后，启动消息接收者，输入消息体。
[root@kafka-node-001 ~]# /opt/kafka/bin/kafka-console-producer.sh --topic one-topic --bootstrap-server kafka-node-001:9092
>Hello World!

# 使用消息接收者脚本，指定主题名称，消息的来源者，回车。即可接收到消息提供者发送的消息体。
[root@kafka-node-002 ~]# /opt/kafka/bin/kafka-console-consumer.sh --topic one-topic --from-beginning --bootstrap-server kafka-node-001:9092
Hello World! 

节点故障模拟

# 停止 kafka-node3(节点 ID 3)
[root@kafka-node-003 ~]# systemctl stop kafka

# 此时查看主题，会发现副本在 ID 3 的，Isr 只有一个 ID 是可用的。同时 Leader 为 3 的，会切至其它可用的 Isr 节点
[root@kafka-node-001 ~]# /opt/kafka/bin/kafka-topics.sh --describe --bootstrap-server kafka-node-001:9092 --topic one-topic
Topic: one-topic      TopicId: gFMGRZuqRv6y4qXgvdkaFw PartitionCount: 6       ReplicationFactor: 2    Configs: min.insync.replicas=1,segment.bytes=1073741824
        Topic: one-topic      Partition: 0    Leader: 1       Replicas: 3,1   Isr: 1          Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 2    Leader: 2       Replicas: 2,3   Isr: 2          Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 3    Leader: 2       Replicas: 3,2   Isr: 2          Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 4    Leader: 2       Replicas: 2,1   Isr: 2,1        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 5    Leader: 1       Replicas: 1,3   Isr: 1          Elr:    LastKnownElr: 

# 消息测试(发送)
[root@kafka-node-001 ~]# /opt/kafka/bin/kafka-console-producer.sh --topic one-topic --bootstrap-server kafka-node-001:9092
>Hello
# 消息测试(接收)
[root@kafka-node-002 ~]# /opt/kafka/bin/kafka-console-consumer.sh --topic one-topic --from-beginning --bootstrap-server kafka-node-001:9092
Hello
Processed a total of 1 messages
# 实验证明，在一个节点宕机的情况下，集群可以正常进行消息体的发送与接收。

# 启动服务
[root@kafka-node-003 ~]# systemctl start kafka
# Isr 恢复正常
[root@kafka-node-001 ~]# /opt/kafka/bin/kafka-topics.sh --describe --bootstrap-server kafka-node-001:9092 --topic one-topic
Topic: one-topic      TopicId: gFMGRZuqRv6y4qXgvdkaFw PartitionCount: 6       ReplicationFactor: 2    Configs: min.insync.replicas=1,segment.bytes=1073741824
        Topic: one-topic      Partition: 0    Leader: 3       Replicas: 3,1   Isr: 1,3        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 2    Leader: 2       Replicas: 2,3   Isr: 2,3        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 3    Leader: 3       Replicas: 3,2   Isr: 2,3        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 4    Leader: 2       Replicas: 2,1   Isr: 2,1        Elr:    LastKnownElr: 
        Topic: one-topic      Partition: 5    Leader: 1       Replicas: 1,3   Isr: 1,3        Elr:    LastKnownElr: 

赞赏