Kafka之ISR机制的理解

文章目录

Kafka的基本概念
什么是ISR
ISR的维护机制
ISR的作用
ISR相关配置参数
同步过程
示例代码
总结

Kafka中的ISR（In-Sync Replicas同步副本）机制是确保数据高可用性和一致性的核心组件。

Kafka的基本概念

在Kafka中，数据被组织成主题（Topic），每个主题分为多个分区（Partition）。每个分区有多个副本（Replica），这些副本分布在不同的Broker上，以确保数据的冗余和高可用性。

Leader Replica：每个分区有一个领导副本，负责处理所有读写请求。
Follower Replica：其他副本作为追随者，从领导副本中复制数据。

什么是ISR

ISR（In-Sync Replicas）是一个分区副本集合，这些副本被认为是与领导副本保持同步的。具体来说，ISR中的副本是那些能够在一定时间内（由参数replica.lag.time.max.ms指定）将数据同步到与领导副本相同位置的副本。

Kafka根据副本同步的情况，分成了3个集合:

AR (Assigned Replicas) : 包括ISR和OSR
ISR (In-sync Replicas) : 和leader副本保持同步的副本集合，可以被认为是可靠的数据
OSR (Out-Sync Replicas) :和Leader副本同步失效的副本集合

AR=ISR+OSR。

ISR的维护机制

领导副本更新ISR：
领导副本会定期检查每个追随者副本的状态。如果某个追随者副本在指定时间内未能跟上领导副本的更新，领导副本会将其从ISR中移除。
追随者副本重新加入ISR：
当追随者副本追上了领导副本的日志进度（即达到了与领导副本相同的日志偏移量），领导副本会将其重新加入ISR。

ISR的作用

数据一致性保证：
生产者在写入数据时，可以通过设置acks参数来控制数据的一致性级别。设置acks=all（或acks=-1）时，领导副本会等待所有ISR中的副本都确认收到数据后，才向生产者发送确认。这保证了数据在写入时至少被写入到ISR中的所有副本。
故障容错能力：
如果领导副本发生故障，Kafka会从ISR中选取一个新的领导副本。由于ISR中的副本与之前的领导副本保持同步，新的领导副本能够继续提供服务，而不会丢失数据。

ISR相关配置参数

replica.lag.time.max.ms：

追随者副本与领导副本之间的最大允许同步延迟时间。如果追随者副本超过此时间没有同步到领导副本，会被移出ISR。

min.insync.replicas：

最少同步副本数。生产者在设置acks=all时，只有当ISR中的副本数不少于这个值，才会确认消息的写入。这个参数用于在保证数据可用性的同时，控制生产者的写入成功率。

同步过程

a. 数据写入
客户端将数据写入主副本。
主副本将数据写入本地日志并确认写入。
主副本异步将数据推送给所有的从副本。
b. 副本同步
从副本收到主副本的数据后，将其写入本地日志，并返回确认。
当从副本确认写入数据后，主副本会更新其 ISR（In-Sync Replicas，即同步副本集合），表示这些副本已经同步到最新的数据。

示例代码

以下是一个简单的生产者示例代码，展示了如何使用acks参数来确保数据写入的高可用性：

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.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.Callback;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // 确保数据被所有ISR副本确认

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

        try {
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key" + i, "value" + i);
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null) {
                            System.out.println("Message sent successfully to partition " + metadata.partition() + " with offset " + metadata.offset());
                        } else {
                            exception.printStackTrace();
                        }
                    }
                });
            }
        } finally {
            producer.close();
        }
    }
}