IT评测·应用市场-qidao123.com

标题: Kafka简单入门02——ISR机制 [打印本页]

---

作者: 惊雷无声    时间: 2024-7-31 11:19
标题: Kafka简单入门02——ISR机制
目次
ISR机制
ISR 关键概念
HW和LEO
Java利用Kafka通讯
Kafka 生产者示例
Kafka 消耗者示例

---

ISR机制

Kafka 中的 ISR（In-Sync Replicas）机制是一种用于确保数据可靠性和一致性的重要机制。ISR 是一组副本，它包括分区的向导者（Leader）和追随者（Follower）副本，这些副本与向导者保持数据同步。
ISR 关键概念

• 向导者和追随者：每个分区有一个向导者和零个或多个追随者。向导者负责处置惩罚客户端的写请求，而追随者重要用于数据复制。
  
• ISR 集合：ISR 集合是分区向导者的一组追随者副本，它们与向导者保持数据同步。只有在 ISR 集合中的追随者副本可以参与数据的写入和读取操作。
  
• 数据复制：向导者将消息写入其本地日志，并定期将这些消息发送给 ISR 集合中的追随者。追随者接收消息后，将其写入本地日志，以保持数据同步。
  
• Leader Epoch 和 Log Start Offset：ISR 集合中的每个追随者都维护了向导者的日志信息，包括向导者的 Leader Epoch 和 Log Start Offset。这些信息用于确保数据的精确复制和同步。
  
• 数据一致性：只有在 ISR 集合中的所有追随者都成功复制了一条消息后，向导者才会将该消息标记为已提交，确保数据的一致性。
  
• 故障处置惩罚：如果某个追随者发生故障大概追赶进度过慢，那么该追随者可能会被从 ISR 集合中移除。这有助于保持数据的可靠性和避免影响性能。
  

其中，需要留意的的概念：

• 分区中的所有副本统称为AR(Assigned Replicas)。
  
• 所有Leader副本加上和Leader副本保持同步的Follower副本构成ISR(In-Sync Replicas)。
  
• 所有没有保持同步的Follower副本构成OSR(Out-of-Sync Replicas)。
  
• AR = ISR + OSR。正常情况下，所有Follower副本都应该和Leader副本一致，即AR=ISR。
  
• 当Leader故障时，在ISR集合中的Follower才有资格被推举为新的Leader。
  

HW和LEO

在 Kafka 中，HW（High Watermark）和 LEO（Log End Offset）是与数据复制和消耗有关的两个重要概念。
HW（High Watermark）：HW 是指在分区中，已经被所有追随者（Follower）副本复制的消息的位置。HW 是每个分区的属性，它体现已经提交的消息。只有在 HW 之前的消息才被以为是已经提交的，这些消息已经被写入分区的所有追随者副本，而且被以为是安全的，不会丢失。HW 是为了确保数据一致性和可靠性而引入的。
LEO（Log End Offset）：LEO 是指在分区中当前最新消息的位置。LEO 体现分区日志中的末了一条消息的偏移量。LEO 包括已经被写入但尚未被所有追随者副本复制的消息，以及正在等待被写入的消息。LEO 是一个动态的属性，它会随着新消息的写入而渐渐增加。
HW 和 LEO 之间的关系非常重要，它们可以帮助确保数据的可靠性和一致性：

• HW 之前的消息是已经提交的消息，它们在数据复制中是安全的，不会丢失。
  
• LEO 之前的消息是已经写入但尚未被所有追随者副本复制的消息。这些消息可能会在 HW 之前被提交，也可能会在之后被提交。
  
• 一旦 HW 追赶上 LEO，体现所有的消息都已经提交，分区的数据一致性得到了保障。
  

Kafka的消息同步流程：

• 初始状态，HW和LEO在同一位置。消耗者可以读取的有效消息为0,1,2,3.
  
  
  
  
  
• 消息写入Leader，LEO位置改变。Follower进行同步。
  
  
  
  
  
• Follower同步进度决定HW位置，消耗者可读的有效消息0,1,2,3,4。
  
  
  
  
  
• 完成同步，消耗者可读的有效消息0,1,2,3,4,5,6。
  
  
  
  
  

可以看出，Kafka的复制机制既不是完全的同步复制，也不是单纯异步复制。

• 同步复制要求所有Follower副本都复制完，太影响性能了。
  
• 异步复制只要数据被写入Leader副本就以为提交成功，在这种情况下，如果Leader宕机时候Follower还是落后于Leader就会造成数据丢失。
  

而Kafka利用的ISR机制则有效地权衡了数据可靠性和性能之间的关系。
Java利用Kafka通讯

以下是 Kafka 生产者和消耗者的简单示例，利用 Kafka 的 Java 客户端库（Kafka Producer 和 Kafka Consumer）来创建一个基本的消息通报示例。
Kafka 生产者示例

1. import org.apache.kafka.clients.producer.*;
2. import java.util.Properties;
3. ​
4. public class KafkaProducerExample {
5.    public static void main(String[] args) {
6.        String bootstrapServers = "localhost:9092"; // Kafka 服务器地址
7.        String topic = "my-topic"; // Kafka 主题名称
8. ​
9.        Properties properties = new Properties();
10.        properties.put("bootstrap.servers", bootstrapServers);
11.        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
12.        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
13. ​
14.        Producer<String, String> producer = new KafkaProducer<>(properties);
15. ​
16.        // 发送消息
17.        producer.send(new ProducerRecord<>(topic, "key", "Hello, Kafka!"), (metadata, exception) -> {
18.            if (exception == null) {
19.                System.out.println("Message sent to partition " + metadata.partition() + ", offset " + metadata.offset());
20.            } else {
21.                System.err.println("Error sending message: " + exception.getMessage());
22.            }
23.        });
24. ​
25.        producer.close();
26.    }
27. }

复制代码

Kafka 消耗者示例

1. import org.apache.kafka.clients.consumer.*;
2. import java.util.Properties;
3. import java.time.Duration;
4. import java.util.Collections;
5. ​
6. public class KafkaConsumerExample {
7.    public static void main(String[] args) {
8.        String bootstrapServers = "localhost:9092"; // Kafka 服务器地址
9.        String groupId = "my-group"; // 消费者组 ID
10.        String topic = "my-topic"; // Kafka 主题名称
11. ​
12.        Properties properties = new Properties();
13.        properties.put("bootstrap.servers", bootstrapServers);
14.        properties.put("group.id", groupId);
15.        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
16.        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
17. ​
18.        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
19.        consumer.subscribe(Collections.singletonList(topic));
20. ​
21.        while (true) {
22.            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
23.            for (ConsumerRecord<String, String> record : records) {
24.                System.out.println("Received message: key = " + record.key() + ", value = " + record.value());
25.            }
26.        }
27.    }
28. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/)

Powered by Discuz! X3.4