Kafka ISR 与 AR 深度解析：副本同步机制核心概念

2.2 关键参数

# Kafka Broker 配置
# 1. 副本落后最大时间阈值
replica.lag.time.max.ms=30000 # 默认 30 秒
# 如果 Follower 超过 30 秒没有拉取消息，就会被踢出 ISR

# 2. 最小同步副本数
min.insync.replicas=2 # Topic 级别或 Broker 级别配置
# 生产者在 acks=all 时需要 ISR 中至少有这么多个副本

2.3 ISR 的动态变化

Kafka ISR 动态变化示例：

1. 正常状态：Leader 0，ISR: [0, 1, 2, 3, 4]
2. Follower 变慢：Follower 4 网络延迟/GC/负载高，同步落后
3. 从 ISR 中移除：超过 replica.lag.time.max.ms 阈值，Follower 4 被踢出
4. Follower 恢复：重新同步，追上 Leader 的 offset，重新加入 ISR

三、ISR 的核心作用

3.1 作用一：Leader 选举的选民池

当 Leader 宕机时，Controller 检测到故障并开始选举。

• 检查 ISR 列表：当前 ISR: [1, 2, 3]
• 选举规则：优先从 ISR 中选择，确保数据不丢失
• 特殊情况：ISR 为空时启用脏选举（unclean leader election），可能丢失数据

为什么只能用 ISR 选举？

• ISR 中的副本数据与 Leader 基本一致
• 选它们做 Leader 不会丢失数据
• 不在 ISR 中的副本数据落后，当选会导致数据丢失

3.2 作用二：消息确认的基准

// Producer 的 acks 参数
Properties props = new Properties();
props.put("acks", "0"); // 可能丢数据，性能最高
props.put("acks", "1"); // Leader 写入成功就返回
props.put("acks", "all"); // 最安全，性能最低
// 或 acks=-1（等同于 all）

acks=all 的工作原理：

• 消息需写入本地并同步给所有 ISR 副本
• 等待 min.insync.replicas 个确认即可返回成功
• 若同步慢，可能导致超时

3.3 作用三：保证数据一致性

// 结合 min.insync.replicas 保证最小同步数
// Topic 配置
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--alter --topic my-topic --config min.insync.replicas=2

// Producer 配置
props.put("acks", "all"); // 要求所有 ISR 确认
// 当 ISR 数量小于 min.insync.replicas 时，Producer 会收到异常

四、AR 与 ISR 的关系图解

4.1 集合关系图

• AR: 所有副本 (Leader + Follower 1, 2, 3...)
• ISR: 同步副本集合 (Leader + 同步 Follower)
• OSR: 不同步副本 (AR - ISR)

4.2 公式表示

AR = {Leader, Follower1, Follower2, Follower3, ...}
ISR = {Leader, Follower_i, Follower_j, ...} 其中 Follower 满足同步条件
OSR = AR - ISR # Out-of-Sync Replicas，同步落后的副本
AR = ISR ∪ OSR
ISR ∩ OSR = ∅

4.3 动态变化示例

# 查看副本状态
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--describe --topic my-topic

# 输出示例
Topic: my-topic Partition: 0 Leader: 0 Replicas: 0,1,2,3,4 Isr: 0,1,2,3
# AR: [0,1,2,3,4]
# ISR: [0,1,2,3]
# OSR: [4]

五、ISR 相关的关键配置

5.1 Broker 级别配置

# server.properties
# Follower 落后多长时间被踢出 ISR（默认 30 秒）
replica.lag.time.max.ms=30000

# 是否允许不在 ISR 中的副本被选举为 Leader
# 如果设为 true，可能会丢数据；设为 false，提高数据一致性
unclean.leader.election.enable=false

# 最小同步副本数（也可在 Topic 级别设置）
min.insync.replicas=1

5.2 Topic 级别配置

# 创建 Topic 时设置
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--create --topic my-topic --partitions 3 --replication-factor 3 \
--config min.insync.replicas=2

# 修改现有 Topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--alter --topic my-topic --config min.insync.replicas=2

5.3 Producer 级别配置

// Producer 配置影响 ISR 的使用
Properties props = new Properties();

// 一致性要求高的场景
props.put("acks", "all"); // 等待所有 ISR 确认
props.put("retries", "3"); // 重试次数
props.put("max.in.flight.requests.per.connection", "1"); // 保证顺序

// 性能优先的场景
props.put("acks", "1"); // Leader 确认即可
props.put("retries", "0"); // 不重试

六、ISR 的实际应用场景

6.1 场景一：Leader 选举

// 当 Leader 宕机时，Kafka 的选举逻辑
public class LeaderElection {
    public void electNewLeader(Partition partition) {
        // 1. 获取当前 ISR 列表
        List<Replica> isr = partition.getInSyncReplicas();
        // 2. 优先从 ISR 中选举
        if (!isr.isEmpty()) {
            // 选择 ISR 中的第一个作为新 Leader
            Replica newLeader = isr.get(0);
            partition.setLeader(newLeader);
        } else {
            // 如果 ISR 为空，根据配置决定
            if (uncleanLeaderElectionEnabled) {
                // 允许从 OSR 中选举（可能丢数据）
                Replica newLeader = chooseFromOSR();
                partition.setLeader(newLeader);
            } else {
                throw new NoReplicaAvailableException();
            }
        }
    }
}

6.2 场景二：生产者的可靠性保证

// 需要高可靠性的生产者配置
public class ReliableProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", Integer.MAX_VALUE);
        props.put("max.in.flight.requests.per.connection", "1");
        props.put("enable.idempotence", "true"); // 开启幂等性
        // 还需要 Topic 配置 min.insync.replicas=2

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("my-topic", "key", "value"), (metadata, exception) -> {
            if (exception != null) {
                // 如果 ISR 不足，会收到 NotEnoughReplicasException
                handleException(exception);
            }
        });
    }
}

6.3 场景三：监控 ISR 变化

# 监控 ISR 收缩和扩张
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--describe --topic my-topic --under-replicated-partitions

# 查看所有未完全同步的分区
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--describe --under-replicated-partitions

七、面试高频问题

Q1：ISR 和 AR 分别代表什么？

答：

• AR：分区的所有副本集合（包括 Leader 和所有 Follower）
• ISR：与 Leader 保持同步的副本集合（包括 Leader 和同步的 Follower）

Q2：副本被踢出 ISR 的条件是什么？

答：Follower 超过 replica.lag.time.max.ms（默认 30 秒）没有向 Leader 拉取消息，就会被踢出 ISR。这表示该 Follower 已经落后太多，不能保证数据一致性。

Q3：为什么 Leader 选举只能从 ISR 中选？

答：因为 ISR 中的副本数据与 Leader 基本一致，选它们做 Leader 不会丢失数据。如果从 OSR 中选举，选出的 Leader 可能缺少大量数据，导致数据不一致。

Q4：min.insync.replicas 的作用是什么？

答：当 Producer 设置 acks=all 时，消息必须写入 ISR 中至少 min.insync.replicas 个副本才算成功。这个参数保证了最小的数据冗余度，防止在 ISR 数量不足时仍然写入成功导致数据风险。

Q5：什么是 unclean leader election？有什么风险？

答：当 ISR 为空时，是否允许从 OSR（不同步副本）中选举 Leader。如果允许（unclean.leader.election.enable=true），可能会选出数据落后的副本作为 Leader，导致数据丢失。这是用一致性换可用性。

八、总结

8.1 核心概念关系

• AR：所有副本集合
• ISR：同步副本集合（可被选举）
• OSR：落后副本集合（不可被选举）

8.2 关键公式

AR = ISR ∪ OSR
ISR ⊆ AR
Leader ∈ ISR

8.3 一句话总结

ISR 是 Kafka 保证数据一致性的核心机制，它圈定了'可用且可靠'的副本集合，是 Leader 选举和数据确认的基准。