Kafka之分区和副本详解_kafka集群分区和副本的分布

Kafka0.8版本后加入副本机制，每个Partition可能有多个备份，某个Partition的Replica列表叫作AR（Assigned Replicas）

1、优先副本（Preferred Replica）

如果一个分区有3个副本，且这3个副本的优先级别分别为1，5，9，根据优先副本的概念，1会作为leader。

AR中的第一个Replica即为“Preferred Replica”。创建一个新的Topic或者给已有Topic增加Partition时，Kafka保证Preferred Replica被均匀分布到集群中的所有Broker上。理想情况下，Preferred Replica会被选为Leader。以上两点保证了所有Partition的Leader被均匀分布到了集群当中，这一点非常重要，因为所有的读写操作都由Leader完成，若Leader分布过于集中，会造成集群负载不均衡。但是，随着集群的运行，该平衡可能会因为Broker的宕机而被打破。

事实上，每个Leader从失败中恢复过来后，它默认会被设置为Follower角色，除非某个Partition的Replica全部宕机，而当前Broker是该Partition的AR中第一个恢复回来的Replica。因此，某个Partition的Leader（Preferred Replica）宕机并恢复后，它很可能不再是该Partition的Leader，但仍然是Preferred Replica。

2、程序指定分区数和复制因子

Producer程序里指定topic，如果topic在broker里不存在，Producer将会通过读取server.properties文件中的两个配置创建它。

两个配置如下：

num.partitions

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1

default.replication.factor

public static void main(String[] args) {
    ZkUtils zkUtils = ZkUtils.
            apply("master:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
 
    AdminUtils.createTopic(zkUtils, "javaCreateTestTopic",  2,
            2,  new Properties(), new RackAwareMode.Enforced$());
    zkUtils.close();
}

查看创建后的topic详情：

Topic:javaCreateTestTopic    PartitionCount:2    ReplicationFactor:2    Configs:
    Topic: javaCreateTestTopic    Partition: 0    Leader: 0    Replicas: 0,1 Isr: 0,1
    Topic: javaCreateTestTopic    Partition: 1    Leader: 1    Replicas: 1,2 Isr: 1,2

3、Partition复制机制

Kafka 主题中的每个分区都有一个预写日志（write-ahead log），写入 Kafka 的消息就存储在这里面。这里面的每条消息都有一个唯一的偏移量，用于标识它在当前分区日志中的位置。

Kafka 中的每个主题分区都被复制了 n 次，其中的 n 是主题的复制因子（replication factor）。这允许 Kafka 在集群服务器发生故障时自动切换到这些副本，以便在出现故障时消息仍然可用。Kafka 的复制是以分区为粒度的，分区的预写日志被复制到 n 个服务器。 在 n 个副本中，一个副本作为 leader，其他副本成为 followers。顾名思义，producer 只能往 leader 分区上写数据（读也只能从 leader 分区上进行），follower会被动定期地去复制leader上的数据。。

日志复制算法（log replication algorithm）必须提供的基本保证是，如果它告诉客户端消息已被提交，而当前 leader 出现故障，新选出的 leader 也必须具有该消息。在出现故障时，Kafka 会从挂掉 leader 的 ISR 里面选择一个 follower 作为这个分区新的 leader ；换句话说，是因为这个 follower 是跟上 leader 写进度的。

每个分区的 leader 会维护一个 in-sync replica（同步副本列表，又称 ISR）。当 producer 往 broker 发送消息，消息先写入到对应 leader 分区上，然后复制到这个分区的所有副本中。只有将消息成功复制到所有同步副本（ISR）后，这条消息才算被提交。由于消息复制延迟受到最慢同步副本的限制，因此快速检测慢副本并将其从 ISR 中删除非常重要。 Kafka 复制协议的细节会有些细微差别。

（1）副本在什么情况下才算跟上 leader

一个副本如果它没有跟上 leader 的日志进度，那么它可能会被标记为不同步的副本。通过一个例子来解释跟上（caught up）的含义。假设我们有一个名为 foo 的主题，并且只有一个分区，同时复制因子为 3。假设此分区的副本分别在 brokers 1，2和3上，并且我们已经在主题 foo 上提交了3条消息。brokers 1上的副本是 leader，副本2和3是 followers，所有副本都是 ISR 的一部分。假设 replica.lag.max.messages 设置为4，这意味着只要 follower 落后 leader 的消息不超过3条，它就不会从 ISR 中删除。我们把 replica.lag.time.max.ms 设置为500毫秒，这意味着只要 follower 每隔500毫秒或更早地向 leader 发送一个 fetch 请求，它们就不会被标记为死亡并且不会从 ISR 中删除。

现在假设 producer 往 leader 上发送下一条消息，与此同时，broker 3 上发生了 GC 停顿

由于 broker 3 在 ISR中，因此在将 broker 3从 ISR 中移除或 broker 3 上的分区跟上 leader 的日志结束偏移之前，最新消息都是不认为被提交的。注意，由于 border 3 落后 leader 的消息比 replica.lag.max.messages = 4 要小，因此不符合从 ISR 中删除的条件。这意味着 broker 3 上的分区需要从 leader 上同步 offset 为 3 的消息，如果它做到了，那么这个副本就是跟上 leader 的。假设 broker 3 在 100ms 内 GC 完成了，并且跟上了 leader 的日志结束偏移，如下图：

（2）什么情况下会导致一个副本与 leader 失去同步

一个副本与 leader 失去同步的原因有很多，主要包括

1）慢副本（Slow replica）：follower replica 在一段时间内一直无法赶上 leader 的写进度。造成这种情况的最常见原因之一是 follower replica 上的 I/O瓶颈，导致它持久化日志的时间比它从 leader 消费消息的时间要长；

2）卡住副本（Stuck replica）：follower replica 在很长一段时间内停止从 leader 获取消息。这可能是以为 GC 停顿，或者副本出现故障；

3）刚启动副本（Bootstrapping replica）：当用户给某个主题增加副本因子时，新的 follower replicas 是不同步的，直到它跟上 leader 的日志。

当副本落后于 leader 分区时，这个副本被认为是不同步或滞后的。在 Kafka 0.8.2 中，副本的滞后于 leader 是根据 replica.lag.max.messages 或 replica.lag.time.max.ms 来衡量的； 前者用于检测慢副本（Slow replica），而后者用于检测卡住副本（Stuck replica）。

（3）如何确认某个副本处于滞后状态

通过 replica.lag.time.max.ms 来检测卡住副本（Stuck replica）在所有情况下都能很好地工作。它跟踪 follower 副本没有向 leader 发送获取请求的时间，通过这个可以推断 follower 是否正常。另一方面，使用消息数量检测不同步慢副本（Slow replica）的模型只有在为单个主题或具有同类流量模式的多个主题设置这些参数时才能很好地工作，但我们发现它不能扩展到生产集群中所有主题。

在之前的示例的基础上，如果主题 foo 以 2 msg/sec 的速率写入数据，其中 leader 收到的单个批次通常永远不会超过3条消息，那么我们知道这个主题的 replica.lag.max.messages 参数可以设置为 4。为什么？ 因为我们以最大速度往 leader 写数据并且在 follower 副本复制这些消息之前，follower 的日志落后于 leader 不超过3条消息。同时，如果主题 foo 的 follower 副本始终落后于 leader 超过3条消息，则希望 leader 删除慢速 follower 副本以防止消息写入延迟增加。

这本质上是 replica.lag.max.messages 的目标 - 能够检测始终与 leader 不同步的副本。假设现在这个主题的流量由于峰值而增加，生产者最终往 foo 发送了一批包含4条消息，等于 replica.lag.max.messages = 4 的配置值。此时，两个 follower 副本将被视为与 leader 不同步，并被移除 ISR。

但是，由于两个 follower 副本都处于活动状态，因此它们将在下一个 fetch 请求中赶上 leader 的日志结束偏移量并被添加回 ISR。如果生产者继续向 leader 发送大量的消息，则将重复上述相同的过程。这证明了 follower 副本进出 ISR 时触发不必要的错误警报的情况。

replica.lag.max.messages 参数的核心问题是，用户必须猜测如何配置这个值，因为不知道 Kafka 的传入流量到底会到多少，特别是在网络峰值的情况下。

（4）参数设定

检测卡住或慢速副本真正重要的事情，是副本与 leader 不同步的时间。删除了通过猜测来设置的 replica.lag.max.messages 参数。现在，只需要在服务器上配置 replica.lag.time.max.ms 参数即可；这个参数的含义为副本与 leader 不同步的时间。

1）检测卡住副本（Stuck replica）的方式与以前相同 - 如果副本未能在 replica.lag.time.max.ms 时间内发送 fetch 请求，则会将其视为已死的副本并从 ISR 中删除；

2）检测慢副本的机制已经改变 - 如果副本落后于 leader 的时间超过 replica.lag.time.max.ms，则认为它太慢并且从 ISR 中删除。

即使在峰值流量下，生产者往 leader 发送大量的消息，除非副本始终和 leader 保持 replica.lag.time.max.ms 时间的落后，否则它不会随机进出 ISR。

注意：

Kafka 0.9.0.0版本后移除了replica.lag.max.messages参数，只保留了replica.lag.time.max.ms作为ISR中副本管理的参数。为什么这样做呢？replica.lag.max.messages表示当前某个副本落后leader的消息数量超过了这个参数的值，那么leader就会把follower从ISR中删除。假设设置replica.lag.max.messages=4，那么如果producer一次传送至broker的消息数量都小于4条时，因为在leader接受到producer发送的消息之后而follower副本开始拉取这些消息之前，follower落后leader的消息数不会超过4条消息，故此没有follower移出ISR，所以这时候replica.lag.max.message的设置似乎是合理的。但是producer发起瞬时高峰流量，producer一次发送的消息超过4条时，也就是超过replica.lag.max.messages，此时follower都会被认为是与leader副本不同步了，从而被踢出了ISR。但实际上这些follower都是存活状态的且没有性能问题。那么在之后追上leader,并被重新加入了ISR。于是就会出现它们不断地剔出ISR然后重新回归ISR，这无疑增加了无谓的性能损耗。而且这个参数是broker全局的。设置太大了，影响真正“落后”follower的移除；设置的太小了，导致follower的频繁进出。无法给定一个合适的replica.lag.max.messages的值，故此，新版本的Kafka移除了这个参数。

4、Leader选举

一条消息只有被ISR中的所有follower都从leader复制过去才会被认为已提交。这样就避免了部分数据被写进了leader，还没来得及被任何follower复制就宕机了，而造成数据丢失。而对于producer而言，它可以选择是否等待消息commit，这可以通过request.required.acks来设置。这种机制确保了只要ISR中有一个或者以上的follower，一条被commit的消息就不会丢失。

有一个很重要的问题是当leader宕机了，怎样在follower中选举出新的leader，因为follower可能落后很多或者直接crash了，所以必须确保选择“最新”的follower作为新的leader。一个基本的原则就是，如果leader不在了，新的leader必须拥有原来的leader commit的所有消息。这就需要做一个折中，如果leader在一个消息被commit前等待更多的follower确认，那么在它挂掉之后就有更多的follower可以成为新的leader，但这也会造成吞吐率的下降。

一种非常常用的选举leader的方式是“少数服从多数”，Kafka并不是采用这种方式。这种模式下，如果我们有2f+1个副本，那么在commit之前必须保证有f+1个replica复制完消息，同时为了保证能正确选举出新的leader，失败的副本数不能超过f个。这种方式有个很大的优势，系统的延迟取决于最快的几台机器，也就是说比如副本数为3，那么延迟就取决于最快的那个follower而不是最慢的那个。“少数服从多数”的方式也有一些劣势，为了保证leader选举的正常进行，它所能容忍的失败的follower数比较少，如果要容忍1个follower挂掉，那么至少要3个以上的副本，如果要容忍2个follower挂掉，必须要有5个以上的副本。也就是说，在生产环境下为了保证较高的容错率，必须要有大量的副本，而大量的副本又会在大数据量下导致性能的急剧下降。这种算法更多用在Zookeeper这种共享集群配置的系统中而很少在需要大量数据的系统中使用的原因。HDFS的HA功能也是基于“少数服从多数”的方式，但是其数据存储并不是采用这样的方式。
Kafka在Zookeeper中为每一个partition动态的维护了一个ISR，这个ISR里的所有replica都跟上了leader，只有ISR里的成员才能有被选为leader的可能（unclean.leader.election.enable=false）。在这种模式下，对于f+1个副本，一个Kafka topic能在保证不丢失已经commit消息的前提下容忍f个副本的失败，在大多数使用场景下，这种模式是十分有利的。事实上，为了容忍f个副本的失败，“少数服从多数”的方式和ISR在commit前需要等待的副本的数量是一样的，但是ISR需要的总的副本的个数几乎是“少数服从多数”的方式的一半。

在ISR中至少有一个follower时，Kafka可以确保已经commit的数据不丢失，但如果某一个partition的所有replica都挂了，就无法保证数据不丢失了。这种情况下有两种可行的方案：

（1）等待ISR中任意一个replica“活”过来，并且选它作为leader（unclean.leader.election.enable=false）

（2）选择第一个“活”过来的replica（并不一定是在ISR中）作为leader（有可能少数据，unclean.leader.election.enable=true）

如果一定要等待ISR中的replica“活”过来，那不可用的时间就可能会相对较长。默认情况下，Kafka采用第二种策略，即unclean.leader.election.enable=true。

用一个实例说明：

假设某个partition中的副本数为3，replica-0, replica-1, replica-2分别存放在broker0, broker1和broker2中。AR=(0,1,2)，ISR=(0,1)。设置request.required.acks=-1, min.insync.replicas=2，unclean.leader.election.enable=false。这里将broker0中的副本也称之为broker0起初broker0为leader，broker1为follower。

当ISR中的replica-0出现crash的情况时，broker1选举为新的leader[ISR=(1)]，因为受min.insync.replicas=2影响，write不能服务，但是read能继续正常服务。

此种情况恢复方案：

（1）尝试恢复(重启)replica-0，如果能起来，系统正常；
（2）如果replica-0不能恢复，需要将min.insync.replicas设置为1，恢复write功能。

当ISR中的replica-0出现crash，紧接着replica-1也出现了crash, 此时[ISR=(1),leader=-1],不能对外提供服务，

此种情况恢复方案：

（1）尝试恢复replica-0和replica-1，如果都能起来，则系统恢复正常；
（2）如果replica-0起来，而replica-1不能起来，这时候仍然不能选出leader，因为当设置unclean.leader.election.enable=false时，leader只能从ISR中选举，当ISR中所有副本都失效之后，需要ISR中最后失效的那个副本能恢复之后才能选举leader, 即replica-0先失效，replica-1后失效，需要replica-1恢复后才能选举leader。保守的方案建议把unclean.leader.election.enable设置为true,但是这样会有丢失数据的情况发生，这样可以恢复read服务。同样需要将min.insync.replicas设置为1，恢复write功能；
（3）replica-1恢复，replica-0不能恢复，这个情况上面遇到过，read服务可用，需要将min.insync.replicas设置为1，恢复write功能；
（4）replica-0和replica-1都不能恢复，这种情况可以参考情形（2）

当ISR中的replica-0, replica-1同时宕机,此时[ISR=(0,1)],不能对外提供服务，

此种情况恢复方案：

（1）尝试恢复replica-0和replica-1，当其中任意一个副本恢复正常时，对外可以提供read服务。直到2个副本恢复正常，write功能才能恢复，或者将将min.insync.replicas设置为1。