使用 Kafka 设计容错微服务

2025 年 5 月 15 日 | 阅读 11 分钟

容错微服务旨在确保系统在部分故障的情况下仍能保持可用。其核心思想是，故障是不可避免的，但其对系统整体功能的影响可以被最小化。

为什么需要容错？

考虑一个音乐流媒体平台，歌曲推荐依赖于一系列服务，例如：

数据摄入：捕获用户交互。
分析：处理用户数据以提供推荐。
交付：向用户提供推荐。

如果其中任何一个服务发生故障，整个推荐系统可能会崩溃。容错可确保单个组件的故障得到隔离，不会导致整个系统宕机。

Kafka 的作用

Kafka 提供以下功能：

异步通信：微服务无需相互等待。
解耦：生产者和消费者通过 Kafka 主题进行交互。
持久性：事件被持久存储。
弹性：副本机制确保即使在 Broker 发生故障时数据也可用。

Kafka 中用于容错的关键组件

理解 Kafka 的构建模块对于设计容错系统至关重要。

a. 主题 (Topics) 和分区 (Partitions)

主题是消息发布的逻辑通道。Kafka 中的每个主题都分为更小的部分，称为分区。这有助于跨多个工作进程同时处理数据，并能防止故障。

示例程序：创建带副本的主题

# Create a topic with 3 partitions and replication factor of 3
kafka-topics.sh --create \
--bootstrap-server localhost:9092 \
--replication-factor 3 \
--partitions 3 \
--topic fault-tolerant-topic

b. 生产者 (Producers) 和消费者 (Consumers)

生产者将数据发送到主题，消费者从中检索数据。两者都需要处理错误以确保可靠性。

生产者示例：可靠消息传递

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class ReliableProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all"); // Ensures all replicas acknowledge
        props.put("retries", 3); // Retry on failure
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props);
        try {
            producer.send(new ProducerRecord<>("fault-tolerant-topic", "key", "value"), (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Error sending message: " + exception.getMessage());
                } else {
                    System.out.println("Message sent to partition " + metadata.partition());
                }
            });
        } finally {
            producer.close();
        }
    }
}

输出

Designing Fault-Tolerant Microservices with Kafka

消费者示例：可靠消费

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class ReliableConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "fault-tolerant-group");
        props.put("enable.auto.commit", "false"); // Manual offset control
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

        try {
            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%n", record.key(), record.value());
                    consumer.commitSync(); // Manually commit offsets
                }
            }
        } finally {
            consumer.close();
        }
    }
}

输出

c. Kafka Broker

Kafka Broker 负责存储和提供主题数据。集群中的多个 Broker 可确保高可用性。

程序：Broker 配置示例

在 server.properties 中

broker.id=1
log.dirs=/tmp/kafka-logs
num.network.threads=3
num.io.threads=8
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
zookeeper.connect=localhost:2181

启动 Broker

设计容错微服务

为了理解使用 Kafka 设计容错微服务，我们将通过完整的程序详细探讨以下概念：

使用 Kafka 解耦服务
事件溯源 (Event Sourcing) 和 CQRS
幂等性

每个概念都将通过用例和完整的 Java 实现进行详细阐述。

1. 使用 Kafka 解耦服务

解耦是微服务架构中的一个关键原则。Kafka充当服务之间的中间人，允许它们异步通信。这意味着如果一个服务宕机，其他服务可以继续运行而不会中断。

用例：订单处理

订单服务：创建新订单时会产生一个事件。
库存服务：消费该事件并更新库存。

实施

订单服务：生成事件

每当下新订单时，OrderService 都会向 orders Kafka 主题发布一个事件。

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class OrderService {
    private static final String TOPIC = "orders";
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");
        Producer<String, String> producer = new KafkaProducer<>(props);
        String orderId = "orderId123";
        String orderDetails = "{\"item\": \"Laptop\", \"quantity\": 1}";
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, orderId, orderDetails);
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Order event sent successfully to topic " + metadata.topic() + " partition " + metadata.partition());
            } else {
                System.err.println("Error sending order event: " + exception.getMessage());
            }
        });
        producer.close();
    }
}

输出

库存服务：消费事件

InventoryService 订阅 orders 主题，处理消息并更新库存。

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class InventoryService {
    private static final String TOPIC = "orders";
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "inventory-service");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Processing order: key=%s, value=%s%n", record.key(), record.value());
                    System.out.println("Inventory updated for order " + record.key());
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }
}

输出

流程

OrderService 向 orders Kafka 主题生成事件。
InventoryService 消费这些事件并更新库存。

2. 事件溯源 (Event Sourcing) 和 CQRS

事件溯源将系统中的每个更改存储为事件，从而能够实现状态转换的完整历史记录。CQRS（命令查询职责分离）将读写操作分开以实现可扩展性。

用例：用户账户管理

命令服务：捕获用户操作并将它们存储为事件。
查询服务：读取聚合数据用于报告。

实施

命令服务：生成事件

CommandService 将所有用户操作发布到 user-events 主题。

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class CommandService {
    private static final String TOPIC = "user-events";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");

        Producer<String, String> producer = new KafkaProducer<>(props);

        String userId = "user123";
        String userEvent = "{\"action\": \"LOGIN\", \"timestamp\": \"2024-11-27T10:00:00Z\"}";

        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, userId, userEvent);
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("User event sent to topic " + metadata.topic() + " partition " + metadata.partition());
            } else {
                System.err.println("Error sending user event: " + exception.getMessage());
            }
        });

        producer.close();
    }
}

输出

查询服务：读取聚合数据

QueryService 聚合用户操作以进行分析，例如识别登录趋势。

import org.apache.kafka.clients.consumer.*;

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

public class QueryService {
    private static final String TOPIC = "user-events";
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "query-service");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Aggregating event: key=%s, value=%s%n", record.key(), record.value());
                }
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }
}

输出

流程

CommandService 捕获用户操作并将其存储为事件。
QueryService 消费这些事件以进行分析。

3. 幂等性 (Idempotency)

幂等性确保多次处理同一消息会产生相同的结果。这对于涉及重试的容错至关重要。

用例：支付处理

即使支付事件被多次接收，支付也必须只处理一次。

实施

带唯一 ID 的生产者

为每个支付事件生成一个唯一 ID，以确保幂等性。

带幂等性检查的消费者

消费者维护一个已处理支付 ID 的记录。

import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.*;

public class PaymentService {
    private static final String TOPIC = "payments";
    private static final Set<String> processedPayments = new HashSet<>();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "payment-service");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    if (!processedPayments.contains(record.key())) {
                        System.out.printf("Processing payment: key=%s, value=%s%n", record.key(), record.value());
                        processedPayments.add(record.key());
                        // Payment processing logic here
                    } else {
                        System.out.printf("Duplicate payment ignored: key=%s%n", record.key());
                    }
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }
}

输出 1

输出 2

输出3

流程

为支付消息分配唯一 ID。
跟踪已处理的支付 ID 以避免重复。

Kafka Streams 中容错的关键机制

状态存储 (State Stores) 和变更日志 (Changelogs)
- Kafka Streams 使用本地状态存储进行有状态操作。
- 这些状态存储由 Kafka 中的变更日志支持，从而可以在故障后进行恢复。
检查点 (Checkpointing) 和偏移量跟踪 (Offset Tracking)
- Kafka Streams 定期将偏移量提交给 Kafka，允许它从最后一个提交的点恢复处理。
任务重新平衡 (Task Rebalancing)
- 在发生故障时，Kafka Streams 会将任务重新分配给健康的实例。
Exactly-Once 语义
- Kafka Streams 使用幂等生产者和...事务性写入.
复制的变更日志
- Kafka 会跨 Broker 复制变更日志以确保持久性。

Kafka Streams 中容错的示例程序

1. 设置简单的 Kafka Streams 应用程序

此程序演示了一个基本的单词计数应用程序，并支持容错。

流配置

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.Stores;
import java.util.Properties;

public class FaultTolerantWordCount {
    public static void main(String[] args) {
        // Stream properties
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, 
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("input-topic");

        KTable<String, Long> wordCounts = textLines
            .flatMapValues(value -> List.of(value.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count(Materialized.<String, Long>as(Stores.persistentKeyValueStore("word-count-store")));

        wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // Add shutdown hook to close the stream gracefully
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

        streams.start();
    }
}

输出

容错的关键特性

状态存储 (word-count-store)
- 由 Kafka 变更日志支持，确保在故障期间状态持久化。
Exactly-Once 处理 (StreamsConfig.EXACTLY_ONCE_V2)
- 保证不丢失或重复数据。
状态目录 (StreamsConfig.STATE_DIR_CONFIG)
- 存储本地状态以加快恢复速度。

2. 从变更日志恢复状态

当 Kafka Streams 应用程序在故障后重新启动时，它会从 Kafka 变更日志主题恢复其状态。此行为可确保不丢失中间状态。

示例：重启时恢复状态

模拟应用程序故障
停止 Kafka Streams 应用程序。
重启应用程序
重启后，Kafka Streams 会...
- 从变更日志主题恢复状态。
- 从最后一个提交的偏移量恢复处理。

3. 处理任务故障

当某个实例发生故障时，Kafka Streams 会动态地将任务重新分配给健康的实例。这种机制可确保持续处理。

示例：扩展 Kafka Streams 应用程序

1. 在多个实例上运行相同的应用程序

java -jar FaultTolerantWordCount.jar &
java -jar FaultTolerantWordCount.jar &

2. 停止一个实例

Kafka Streams 会自动在剩余实例之间重新平衡任务。

4. 实现优雅关机

为了确保容错，Kafka Streams 允许应用程序优雅地关闭，在退出前提交偏移量并持久化状态。

示例：优雅关机钩子

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    System.out.println("Shutting down Kafka Streams...");
    streams.close();
}));

5. Join 操作的容错

Kafka Streams 支持 Join 等有状态操作，由于变更日志，这些操作对故障具有弹性。

示例：流-流 Join

KStream<String, String> orders = builder.stream("orders");
KStream<String, String> shipments = builder.stream("shipments");

KStream<String, String> enrichedOrders = orders.join(
    shipments,
    (orderValue, shipmentValue) -> "Order: " + orderValue + ", Shipment: " + shipmentValue,
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
);

enrichedOrders.to("enriched-orders");

变更日志：为窗口 Join 操作提供状态备份，确保故障后的恢复。
偏移量：在恢复期间从提交的偏移量重新同步。

使用混沌工程在 Kafka 中测试容错

混沌工程是一种分布式系统方法，用于测试其对故障的弹性。在 Kafka 中，测试容错涉及模拟 Broker 崩溃、网络中断和消费者组重新平衡等场景，以确保系统在不利条件下按预期运行。

Kafka 中的容错场景

Broker 崩溃
- 模拟 Broker 离线，以验证 Leader 选举和数据复制。
网络问题
- 引入网络延迟或断开连接，以测试 Kafka 维持操作的能力。
消费者组重新平衡
- 通过添加或删除消费者来触发重新平衡，以评估平滑的任务重新分配。

设置环境

集群配置

Kafka 集群包含 3 个 Broker：broker1、broker2、broker3。

kafka-topics.sh --create \
--topic chaos-testing \
--partitions 3 \
--replication-factor 3 \
--bootstrap-server localhost:9092

生产者代码

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class ChaosProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");

        Producer<String, String> producer = new KafkaProducer<>(props);

        try {
            for (int i = 0; i < 100; i++) {
                producer.send(new ProducerRecord<>("chaos-testing", "key" + i, "value" + i),
                        (metadata, exception) -> {
                            if (exception == null) {
                                System.out.printf("Sent message to topic %s partition %d%n",
                                        metadata.topic(), metadata.partition());
                            } else {
                                System.err.println("Error: " + exception.getMessage());
                            }
                        });
            }
        } finally {
            producer.close();
        }
    }
}

输出

消费者代码

import org.apache.kafka.clients.consumer.*;

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

public class ChaosConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("group.id", "chaos-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("enable.auto.commit", "false");

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("chaos-testing"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed message: key=%s, value=%s, partition=%d%n",
                            record.key(), record.value(), record.partition());
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }
}

输出

使用混沌工程测试容错

1. Broker 崩溃模拟

场景

通过停止集群中的一个 Broker 来模拟 Broker 故障。

步骤：

a. 启动包含 3 个 Broker 的 Kafka 集群。

kafka-server-start.sh config/server.properties --broker.id 1
kafka-server-start.sh config/server.properties --broker.id 2
kafka-server-start.sh config/server.properties --broker.id 3

b. 停止一个 Broker

c. 验证分区的 Leader 选举

kafka-topics.sh --describe --topic chaos-testing --bootstrap-server localhost:9092

d. 观察生产者和消费者的行为

生产者应继续写入剩余的 Broker。
消费者应无缝地从新 Leader 读取数据。

2. 网络分区模拟

场景

引入网络延迟或模拟 Broker 无法访问。

步骤：

a. 使用网络仿真工具（如 tc（Linux 流量控制））引入延迟

b. 临时断开一个 Broker 的连接

c. 监控 Kafka 的行为

检查生产者和消费者是否重试并连接到其他 Broker。
确保没有数据丢失或重复。

d. 恢复网络

3. 消费者组重新平衡模拟

场景

通过在组中添加或删除消费者来模拟任务重新平衡。

步骤：

a. 启动第一个消费者实例

b. 启动其他消费者

java -jar ChaosConsumer.jar &

java -jar ChaosConsumer.jar &
java -jar ChaosConsumer.jar &

c. 观察日志中的重新平衡

消费者之间重新分配分区。

d. 移除一个消费者实例

e. 验证剩余消费者能否无缝接管工作负载。

验证 Kafka 中的容错

1. 数据一致性

为确保数据一致性，请验证 Kafka 的复制机制是否已保存所有消息而没有丢失或重复。通过生产已知数量的消息（`acks` 设置为 "all"），模拟 Broker 故障，然后消费所有消息来测试这一点。消耗的消息数量应与生产的消息数量匹配，并且不应有重复或缺失的数据。

2. 偏移量管理

偏移量管理可确保消费者在故障后从正确的位置恢复。要进行验证，请检查偏移量是否已定期提交到 Kafka（通过消费者日志或 Kafka 的偏移量存储）。在消费者发生故障并重启后，它应从上次中断的地方继续处理，仅处理未消费的记录。

3. 分区 Leader 选举

分区 Leader 选举可确保在 Broker 故障期间的连续性。验证在 Broker 发生故障后，其分区的 Leader 是否已转移到同步副本 (ISR)。使用 Kafka CLI 或管理工具检查分区 Leader 的变更，确保所有分区都有活跃的 Leader，且停机时间最少。

4. 消费者负载均衡

消费者重新平衡可确保在活跃消费者之间平滑地重新分配分区。模拟消费者组的变更（通过添加或删除消费者）并监控系统日志。验证所有分区是否已重新分配给活跃消费者，并且没有处理中断或显著延迟。

下一主题Kafka-and-digital-twin-architectures

使用 Kafka 设计容错微服务

为什么需要容错？

Kafka 的作用

Kafka 中用于容错的关键组件

a. 主题 (Topics) 和分区 (Partitions)

b. 生产者 (Producers) 和消费者 (Consumers)

c. Kafka Broker

设计容错微服务

1. 使用 Kafka 解耦服务

2. 事件溯源 (Event Sourcing) 和 CQRS

3. 幂等性 (Idempotency)

Kafka Streams 中容错的关键机制

1. 设置简单的 Kafka Streams 应用程序

2. 从变更日志恢复状态

3. 处理任务故障

4. 实现优雅关机

5. Join 操作的容错

使用混沌工程在 Kafka 中测试容错

Kafka 中的容错场景

设置环境

使用混沌工程测试容错

验证 Kafka 中的容错

1. 数据一致性

2. 偏移量管理

3. 分区 Leader 选举

4. 消费者负载均衡

联系信息

关注我们

教程

面试题

在线编译器

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 设计容错微服务

为什么需要容错？

Kafka 的作用

Kafka 中用于容错的关键组件

a. 主题 (Topics) 和分区 (Partitions)

b. 生产者 (Producers) 和消费者 (Consumers)

c. Kafka Broker

设计容错微服务

1. 使用 Kafka 解耦服务

2. 事件溯源 (Event Sourcing) 和 CQRS

3. 幂等性 (Idempotency)

Kafka Streams 中容错的关键机制

1. 设置简单的 Kafka Streams 应用程序

2. 从变更日志恢复状态

3. 处理任务故障

4. 实现优雅关机

5. Join 操作的容错

使用混沌工程在 Kafka 中测试容错

Kafka 中的容错场景

设置环境

使用混沌工程测试容错

验证 Kafka 中的容错

1. 数据一致性

2. 偏移量管理

3. 分区 Leader 选举

4. 消费者负载均衡

相关帖子

Kafka Streams API 中的数据流和推荐

使用 Kafka 主题流和 Cosmos DB 进行数据丰富

Apache Kafka vs Apache Storm

Kafka 日志分析

Cosmos DB 作为 Apache Kafka 主题的接收器

针对基因组数据流优化的 Kafka

使用 Kafka Connect 进行边缘计算

使用 Kafka 进行事件驱动的网络安全

ZooKeeper 摘要身份验证

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

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器