Kafka 中的容错性

2025年5月14日 | 阅读12分钟

Apache Kafka 容错简介

在现代分布式计算时代，确保系统在故障条件下的可用性和可靠性至关重要。Apache Kafka，一个分布式事件流平台，因其能够处理海量实时数据流，同时保持高可用性和可靠性（即使在发生故障时）而广受认可。该平台强大的容错机制在实现这一目标方面发挥着至关重要的作用，它允许 Kafka 在硬件、软件或网络故障发生时继续运行。

什么是容错？

在 Kafka 的上下文中，容错确保系统保持可用并且数据不会丢失，即使一个或多个组件（如代理、生产者或消费者）发生故障。这对于分布式系统至关重要，因为由于操作的复杂性和规模，故障是不可避免的。

Kafka 的架构专门用于确保

数据持久性，这样即使在代理或网络故障的情况下也不会丢失任何消息。
高可用性，意味着生产者和消费者即使在发生故障时也能继续发送和接收数据。

Kafka 如何确保容错

1. 数据复制

复制是 Kafka 容错策略的基石。Kafka 通过将数据复制到多个代理来确保高可用性和数据持久性。这样，即使一个代理宕机，数据仍然可以从其他代理访问。

复制因子：复制因子决定了每个分区的副本数量。较高的复制因子提供了更多的备份并增加了容错能力，但它也会消耗更多的资源，如存储和网络带宽。
同步副本 (ISR)：Kafka 使用一个称为同步副本 (ISR) 的系统来跟踪哪些副本与领导者保持最新。如果副本已完全赶上领导者的数据，则该副本被认为是“同步”的。当代理发生故障时，只有 ISR 中的副本才能成为新的领导者，从而确保在此过程中不会丢失任何数据。

2. 领导者选举和故障转移

当当前分区的领导者发生故障时，Kafka 使用Zookeeper（或新版本中的内部 KRaft 控制器）来协调新领导者的选举。此领导者选举过程是自动的，不需要手动干预。只要 ISR 中有健康的跟随者，Kafka 就可以提升其中一个来接任领导者。

这种自动选择新领导者并进行切换的过程可确保即使某些代理离线，系统也能继续运行。

一旦跟随者被提升为领导者，生产者和消费者就可以在没有明显中断的情况下继续发送和接收数据。

3. 生产者和消费者的弹性

Kafka 的生产者和消费者客户端也为容错做出了贡献。生产者可以配置为确保消息在传输过程中不会丢失，即使存在暂时的网络或代理故障。例如，生产者可以在考虑成功发送消息之前等待来自多个副本的确认。

消费者使用偏移量来跟踪他们在数据流中的位置。如果消费者发生故障，Kafka 可确保它可以在最后一个保存的位置重新开始读取，从而降低丢失或重复数据的可能性。

Apache Kafka 容错的关键组成部分

Apache Kafka 的容错通过几个关键组件协同工作来实现，以确保高可用性、可靠性和数据一致性，即使在系统发生故障时也是如此。这些组件包括代理、主题、分区、复制、领导者选举、Zookeeper（或较新的 KRaft 模式）以及消费者偏移量管理。了解这些部分如何协同工作以构建强大可靠的 Kafka 系统非常重要。

在本节中，我们将深入探讨这些组件，并提供实际示例和程序来说明 Kafka 如何实现容错。

1. Kafka 代理

Kafka 代理是存储数据并处理来自生产者和消费者客户端请求的服务器。Kafka 通常设置为一组代理，这有助于其扩展并保持可靠性，即使出现问题。

多个代理可以处理大量数据，如果一个代理发生故障，其他代理可以接管任务以确保服务连续性。

集群中的每个代理都由唯一的 ID 标识，代理共同管理主题的分区和副本。

促进容错的关键特性

集群配置：多个代理协同工作以存储数据并管理分区。
复制：分区中的数据被复制到不同的代理，以避免在一个代理发生故障时丢失数据。

Kafka 代理示例

# Start a Kafka broker using a configuration file
bin/kafka-server-start.sh config/server.properties

您可以通过在不同机器上启动多个代理来扩展 Kafka，从而创建一个容错系统。

2. 主题和分区

在 Kafka 中，主题就像生产者发送消息和消费者接收消息的通道。主题被分成称为分区的较小部分。分区有助于 Kafka 处理更多数据并使其在出现问题时更具弹性。通过将数据分布在不同的分区上，Kafka 可以跨多个代理（服务器）分摊工作负载，这有助于复制。如果一个代理发生故障，拥有该分区副本的其他代理可以继续为用户提供服务。

提高 Kafka 可靠性的关键特性

分区：将消息分成更小的部分（分区）有助于更好地处理数据和提高可靠性。
复制：每个分区都可以复制到其他代理，所以如果一个代理发生故障，系统仍然可以工作。

示例：创建具有分区和复制的主题

# Command to create a Kafka topic with 3 partitions and 3 copies (replicas)
bin/kafka-topics.sh --create --topic fault-tolerant-topic \
    --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092

此命令创建一个名为 fault-tolerant-topic 的主题，包含 3 个分区，每个分区将复制到 3 个代理。这样，如果一个代理停止工作，其他代理可以接管而不会丢失数据。

3. 复制和 ISR（同步副本）

复制是 Kafka 处理故障能力的关键方面。主题中的每个分区都会在多个代理之间进行复制，从而确保数据的多个备份。这种冗余使 Kafka 即使在代理宕机的情况下也能继续运行，而不会丢失任何信息。

领导者：每个分区只有一个代理被指定为领导者。它负责管理该分区的_所有_读写操作。
跟随者：存储分区副本的其他代理称为跟随者。它们通过从领导者复制数据来持续同步它们的数据。
同步副本 (ISR)：Kafka 跟踪一组与领导者完全同步的跟随者。只有 ISR 中的代理才能在当前领导者发生故障时接任领导者。

如果领导者代理发生故障，Kafka 会将一个同步的跟随者提升为新的领导者，从而确保连续性。

示例：设置具有复制的主题

# Create a topic with 4 partitions and a replication factor of 2
bin/kafka-topics.sh --create --topic replicated-topic \
    --partitions 4 --replication-factor 2 --bootstrap-server localhost:9092

此命令设置一个包含 4 个分区的主题，每个分区复制到两个代理。如果一个代理发生故障，剩余的代理将继续提供复制的分区。

查看主题的 ISR

# Describe the topic to view its ISR (In-Sync Replicas)
bin/kafka-topics.sh --describe --topic replicated-topic --bootstrap-server localhost:9092

此命令将提供有关主题的信息，包括其分区、领导者和 ISR，展示 Kafka 如何管理副本以实现容错。

4. 领导者选举和故障转移

Kafka 的领导者选举过程可确保即使在代理发生故障时，分区也能保持可用。当领导者代理不再可用时，Kafka 会自动提升一个同步副本 (ISR) 来接任新的领导者。这种自动故障转移过程对生产者和消费者来说是透明的，他们可以在不中断的情况下继续从新领导者那里读取和写入。

领导者选举由Zookeeper（或 Kafka 较新的KRaft 控制器模式）协调，它会监控代理和分区的运行状况。

领导者选举过程中的步骤

检测故障：Zookeeper 检测到领导者代理已失败。
提升新领导者：Kafka 提升一个同步副本作为新的领导者。
继续操作：新领导者开始处理分区的读写操作，Kafka 更新元数据以反映更改。

示例：模拟代理故障并观察领导者选举

启动两个代理

bin/kafka-server-start.sh config/server-1.properties
bin/kafka-server-start.sh config/server-2.properties

创建具有复制的主题

bin/kafka-topics.sh --create --topic test-topic \
    --partitions 1 --replication-factor 2 --bootstrap-server localhost:9092

现在，停止该分区领导者的代理

检查新的领导者选举

bin/kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092

您将看到一个新的代理已被选为领导者。

5. Zookeeper（或 KRaft 模式）

Zookeeper 负责管理和协调 Kafka 代理、分区和领导者选举。在最近的 Kafka 版本中，引入了KRaft (Kafka Raft) 模式来替代 Zookeeper，它提供了一个集成且更简化的元数据管理解决方案。

Zookeeper 在容错方面的作用

代理协调：Zookeeper 跟踪哪些代理可用且运行正常。
领导者选举：Zookeeper 监控分区，并在领导者代理宕机或失败时启动领导者选举。
配置管理：Zookeeper 存储 Kafka 的元数据，包括分区和副本分配。

Zookeeper 设置示例

首先，您需要启动 Zookeeper

一旦 Zookeeper 运行起来，您就可以启动 Kafka 代理，Zookeeper 将协调集群。

Kafka 较新的KRaft模式消除了对 Zookeeper 的依赖，从而降低了复杂性。KRaft 在 Kafka 代理内部处理所有元数据管理。

以 KRaft 模式启动 Kafka

bin/kafka-storage.sh format --config config/kraft/server.properties --cluster-id <cluster_id>
bin/kafka-server-start.sh config/kraft/server.properties

KRaft 通过减少 Kafka 对外部系统（Zookeeper）的依赖并简化集群管理来提高容错能力。

6. 生产者确认和投递保证

Kafka 生产者将消息发送到 Kafka 主题。Kafka 的生产者 API 通过 acks 参数允许对投递保证进行细粒度控制，该参数控制生产者何时收到来自代理的确认。

生产者确认级别

acks = 0：生产者发送消息，不期望来自代理的任何反馈。此方法速度极快，但不能保证消息已到达目的地。
acks = 1：生产者等待来自领导者代理的确认。如果领导者在将消息复制到其副本之前宕机，则消息可能会丢失。
acks = all：生产者等待所有同步副本 (ISR) 确认消息。这种方法保证消息安全地存储在所有副本中，提供最大程度的数据丢失保护。

配置生产者确认的示例

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");  // Ensure highest fault tolerance
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-fault-tolerant-topic", "key", "value"));
producer.close();

在此示例中，生产者配置为使用 acks=all，这确保在返回确认之前，数据已完全复制到所有同步副本。

7. 消费者偏移量管理

Kafka 中的消费者会跟踪它们已处理消息的偏移量（即位置）。妥善监督客户余额对于实现对内部故障的适应至关重要，因为它允许客户在故障后从中断处继续。

偏移量可以自动或手动提交

自动偏移量提交：Kafka 会定期自动提交偏移量。这更容易使用，但如果在提交偏移量之前发生故障，可能会导致重复消息处理。
手动偏移量提交：消费者在成功处理消息后手动提交偏移量。这可确保即使发生故障，消息也只会被处理一次。

示例：手动提交消费者偏移量

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("enable.auto.commit", "false");  // Disable automatic offset commit
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-fault-tolerant-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // Process the message
        System.out.printf("Consumed message with key: %s, value: %s%n", record.key(), record.value());
    }
    // Manually commit the offset after processing messages
    consumer.commitSync();
}

在此示例中，消费者仅在处理完消息后手动提交偏移量，从而确保在消费者发生故障并重新启动时不会丢失或重新处理数据。

Apache Kafka 容错的监控和告警

监控和告警是维护 Apache Kafka 容错能力的关键组成部分。通过密切关注 Kafka 的健康和性能指标，您可以在代理故障、复制延迟和不干净的领导者选举等问题升级为重大事件之前，快速检测到它们。此外，设置适当的告警可确保在任何威胁 Kafka 可用性或数据完整性的问题发生时，您的团队都能实时收到通知。

本指南探讨了监控 Kafka 以实现容错的关键方面，并提供了使用指标、Prometheus 和Grafana 等工具以及为特定故障场景配置告警的实际示例。

使用 Prometheus 和 Grafana 设置 Kafka 监控

Prometheus 和 Grafana 是用于观测和可视化的著名开源工具。Prometheus 收集和存储指标，而 Grafana 则对其进行可视化，允许您设置仪表板和告警。

1. 设置 Kafka 的 Prometheus JMX Exporter

Kafka 通过 JMX（Java Management Extensions）暴露其指标。要使用 Prometheus 监控 Kafka，您需要一个 JMX 导出器，它是一个轻量级的 Java 代理，通过 HTTP 以 Prometheus 可读的格式公开 JMX 指标。

步骤 1：下载 JMX Exporter 从 Prometheus JMX exporter GitHub 页面下载 JMX exporter JAR。

步骤 2：配置 JMX Exporter 为 JMX 导出器创建一个配置文件 (kafka-jmx-config.yml)

lowercaseOutputName: true
rules:
  - pattern: "kafka.server<type=(.+), name=(.+)>"
    name: "kafka_$1_$2"
    type: GAUGE
    labels:
      cluster: "my-cluster"
  - pattern: "kafka.network<type=(.+), name=(.+)>"
    name: "kafka_network_$1_$2"
    type: GAUGE

步骤 3：运行带有 JMX Exporter 的 Kafka 启动 Kafka 代理时，将 JMX 导出器作为 JVM 代理传递

KAFKA_JMX_OPTS="-javaagent:/path/to/jmx_prometheus_javaagent.jar=7071:/path/to/kafka-jmx-config.yml"
bin/kafka-server-start.sh config/server.properties

此配置使 Kafka 在端口 7071 上公开其指标。

2. 设置 Prometheus

Prometheus 从 Kafka JMX 导出器抓取指标。

步骤 1：安装 Prometheus

步骤 2：配置 Prometheus 编辑 prometheus.yml 文件以将 Kafka 添加为目标

scrape_configs:
  - job_name: 'kafka'
    static_configs:
      - targets: ['localhost:7071']

步骤 3：启动 Prometheus 运行 Prometheus

Prometheus 现在将从 Kafka 代理抓取指标。

3. 设置 Grafana

Grafana 用于可视化存储在 Prometheus 中的指标。

步骤 1：安装 Grafana 从官方 Grafana 网站下载并安装 Grafana。

步骤 2：将 Prometheus 添加为数据源

进入 Grafana UI (https://:3000)。
通过导航到 Configuration > Data Sources > Add data source 来添加 Prometheus 作为数据源。
选择 Prometheus 并配置 URL (https://:9090)。

步骤 3：创建 Kafka 监控仪表板 您可以创建自定义仪表板，或从 Grafana 社区导入现有的 Kafka 仪表板。要创建仪表板

转到 Create > Dashboard 并添加一个新面板。
从 Prometheus 查询指标。例如，要跟踪同步副本的数量

4. Grafana 中要监控的示例 Kafka 指标

代理健康状况：正在运行的代理数量。

副本不足的分区：缺少副本的分区，这表示复制可能存在问题。

同步副本 (ISR)：监控同步副本的数量。

复制延迟：领导者副本和跟随者副本之间的延迟，这对于确保跟随者是最新的是至关重要的。

消费者延迟：最后消费的偏移量与分区中最新偏移量之间的差值。较高的消费者延迟表明消费者正在落后。

为 Kafka 设置告警

没有适当的告警，监控是不完整的。告警会通知您的团队需要立即关注的潜在问题。在 Prometheus 中，使用 alert_rules.yml 文件配置告警，告警管理器负责发送通知（电子邮件、Slack 等）。

步骤 1：在 Prometheus 中定义告警规则

创建一个名为 alert_rules.yml 的文件

groups:
  - name: KafkaFaultToleranceAlerts
    rules:
    - alert: KafkaUnderReplicatedPartitions
      expr: kafka_server_replica_manager_under_replicated_partitions > 0
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "Kafka has under-replicated partitions"
        description: "Under-replicated partitions exist for more than 5 minutes."

    - alert: KafkaOfflinePartitions
      expr: kafka_server_replica_manager_offline_partitions_count > 0
      for: 1m
      labels:
        severity: critical
      annotations:
        summary: "Kafka has offline partitions"
        description: "Offline partitions are detected, indicating a failure in the cluster."

步骤 2：在 Prometheus 中启用告警

在 prometheus.yml 文件中，引用告警规则

rule_files

步骤 3：设置告警管理器

安装并配置 Prometheus 告警管理器以发送通知（例如，通过电子邮件或 Slack）。以下是 Slack 的配置示例

global:
  slack_api_url: 'https://hooks.slack.com/services/TXXXXXX/BXXXXXX/XXXXXX'

route:
  group_by: ['alertname']
  receiver: 'slack-notifications'

receivers:
  - name: 'slack-notifications'
    slack_configs:
    - channel: '#kafka-alerts'
      send_resolved: true

步骤 4：测试告警

停止一个 Kafka 代理以测试您的副本不足告警。Prometheus 将检测到副本不足的分区并触发告警，向配置的 Slack 频道发送通知。

容错 Kafka 生产者和消费者程序

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;
import java.time.Duration;

public class FaultTolerantKafka {

    public static void main(String[] args) {
        // Producer Configuration
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
        producerProps.put(ProducerConfig.RETRIES_CONFIG, 3);

        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

        // Sending messages
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("fault-tolerant-topic", "key" + i, "message" + i);
            producer.send(record, (metadata, exception) -> {
                if (exception == null) {
                    System.out.printf("Produced message: Key = %s, Value = %s, Partition = %d, Offset = %d%n",
                            record.key(), record.value(), metadata.partition(), metadata.offset());
                } else {
                    System.err.println("Error producing message: " + exception);
                }
            });
        }
        producer.close();

        // Consumer Configuration
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        consumerProps.put("group.id", "fault-tolerant-group");
        consumerProps.put("enable.auto.commit", "false");
        consumerProps.put("key.deserializer", StringDeserializer.class.getName());
        consumerProps.put("value.deserializer", StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Collections.singletonList("fault-tolerant-topic"));

        // Consuming messages
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Consumed message: Key = %s, Value = %s, Partition = %d, Offset = %d%n",
                        record.key(), record.value(), record.partition(), record.offset());
                
                // Manual offset commit
                consumer.commitSync();
            }
        }
    }
}

输出

下一主题Kafka 延迟

Kafka 中的容错性

Apache Kafka 容错简介

什么是容错？

Kafka 如何确保容错

1. 数据复制

2. 领导者选举和故障转移

3. 生产者和消费者的弹性

Apache Kafka 容错的关键组成部分

1. Kafka 代理

2. 主题和分区

3. 复制和 ISR（同步副本）

4. 领导者选举和故障转移

5. Zookeeper（或 KRaft 模式）

6. 生产者确认和投递保证

7. 消费者偏移量管理

Apache Kafka 容错的监控和告警

使用 Prometheus 和 Grafana 设置 Kafka 监控

1. 设置 Kafka 的 Prometheus JMX Exporter

2. 设置 Prometheus

容错 Kafka 生产者和消费者程序

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 中的容错性

Apache Kafka 容错简介

什么是容错？

Kafka 如何确保容错

1. 数据复制

2. 领导者选举和故障转移

3. 生产者和消费者的弹性

Apache Kafka 容错的关键组成部分

1. Kafka 代理

2. 主题和分区

3. 复制和 ISR（同步副本）

4. 领导者选举和故障转移

5. Zookeeper（或 KRaft 模式）

6. 生产者确认和投递保证

7. 消费者偏移量管理

Apache Kafka 容错的监控和告警

使用 Prometheus 和 Grafana 设置 Kafka 监控

1. 设置 Kafka 的 Prometheus JMX Exporter

2. 设置 Prometheus

容错 Kafka 生产者和消费者程序

相关帖子

AI 驱动的 Kafka 代理优化

使用 Kafka 的预测性缓存用例

分区策略：Kafka 主题和 Cosmos DB 容器

Kafka Hadoop 加载器

Kafka 延迟

Kafka Connect BigQuery 接收器

Kafka 高级处理流

Kafka 记录

Kafka 消息压缩

节能 Kafka 集群

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器