Kafka ACK机制详解！

创作时间:

作者:

@小白创作中心

Kafka ACK机制详解！

引用

来源

https://developer.aliyun.com/article/1619267

Kafka的ACK机制是确保消息成功传递和处理的重要机制。这篇文章将详细分析Kafka ACK机制，包括其原理、源码分析、使用场景以及优缺点。

ACK 方式

Kafka的ACK机制主要用于确保生产者发送的消息能够被可靠地写入到Kafka集群的Topic中。ACK机制的核心思想是生产者发送消息后，需要等待Kafka集群的确认（ACK），才认为消息发送成功。

Kafka的ACK机制主要有三种级别：

acks=0

生产者不等待服务器的确认，消息发送后即认为成功，不管消息是否真正写入Kafka，这种方式效率最高，但可靠性最低，数据可能存在丢失。

acks=1

生产者会等待来自Leader分区的确认。Leader分区接收到消息并写入本地日志后即返回确认。这种方式在Leader分区可用时可靠，但如果Leader分区发生故障，可能会丢失数据。从Kafka 2.0开始，默认值是acks=1

acks=all（或-1）

生产者等待所有ISR（In-Sync Replica，同步副本）分区的确认。只有当消息被写入所有同步副本后才返回确认，这种方式最可靠，但性能较低。

ISR的工作原理

ISR，全称In-Sync Replicas，翻译为同步副本，它是指某个分区中的一组与Leader副本保持同步的副本，即这些副本包含了Leader副本中的所有已确认消息。ISR是Kafka集群中用于保证数据可靠性的一个关键概念。

Leader和Follower：在Kafka中，每个分区都有一个Leader和若干个Follower，Leader负责处理所有的读写请求，而Follower则从Leader那里拉取数据并进行同步。
同步副本（ISR）：ISR是一个动态的集合，包含了Leader和所有与Leader保持同步的Follower，只有在ISR中的副本才被认为是可靠的，因为它们包含了与Leader相同的数据。
ACK机制与ISR：当生产者发送消息并设置acks=all时，Kafka只有在消息被写入ISR中的所有副本后才会返回确认，这确保了消息即使在Leader故障的情况下也不会丢失，因为ISR中的其他副本可以选举为新的Leader。

ISR的维护

Kafka通过以下机制来维护ISR：

加入ISR：当一个Follower副本成功地追上了Leader副本的日志（即复制了Leader的所有新的消息），它会被加入到ISR中。
移出ISR：当一个Follower副本落后于Leader超过一定的时间（由参数replica.lag.time.max.ms控制），它会被移出ISR。

ISR源码分析

以下是Kafka中维护ISR的关键代码片段（以Kafka 2.x版本为例）：

class Partition {
    private Set<Replica> isr; // 当前分区的ISR集合
    public void updateISR() {
        // 获取所有副本的状态
        List<Replica> replicas = getReplicas();
        // 计算新的ISR集合
        Set<Replica> newIsr = new HashSet<>();
        for (Replica replica : replicas) {
            if (replica.isInSync()) {
                newIsr.add(replica);
            }
        }
        // 更新ISR
        if (!newIsr.equals(this.isr)) {
            this.isr = newIsr;
            // 触发ISR变化的事件
            onISRChanged();
        }
    }
}
class Replica {
    public boolean isInSync() {
        // 判断该副本是否与Leader同步
        return this.logEndOffset >= leaderLogEndOffset - replicaLagMaxMessages;
    }
}

源码分析

以Kafka的Producer端代码为例，下面是简化后的发送消息时处理ACK机制的关键代码片段：

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // 构建请求
    ProduceRequest request = new ProduceRequest(record, callback);
    // 发送请求
    Future<RecordMetadata> future = this.sender.send(request);
    // 根据ACK配置处理确认
    if (this.acks == 0) {
        // 不等待确认，直接返回成功
        callback.onCompletion(null, null);
    } else if (this.acks == 1) {
        // 等待Leader确认
        RecordMetadata metadata = future.get();
        callback.onCompletion(metadata, null);
    } else if (this.acks == -1 || this.acks == "all") {
        // 等待所有ISR确认
        RecordMetadata metadata = future.get();
        callback.onCompletion(metadata, null);
    }
    return future;
}