Kafka集群内复制功能深入剖析

Posted 2021-01-14

Kafka是一个分布式发布订阅消息系统。由LinkedIn开发并已经在2011年7月成为apache顶级项目。kafka在LinkedIn, Twitte等许多公司都得到广泛使用，主要用于：日志聚合，消息队列，实时监控等。

0.8版本开始，kafka支持集群内复制，从而提高可用性和系统稳定性，这篇文章主要概述kafka复制的设计。

复制
有了复制后，kafka客户端将会得到如下好处：

生产者能在出现故障的时候继续发布消息，并且能在延迟和持久性之间选择，取决于应用。
消费者能在出现故障的时候继续实时接受正确的消息。
所有的分布式系统必须在一致性，可用性，分区容错性之间进行权衡并做出取舍(参考CAP定理)，kafka的目标是在单个数据中心里的kafka集群也支持复制。网络分区是比较少见的，因此kafka设计专注于高可用和强一致。强一致意味着所有副本数据完全一致，这简化了应用程序开发人员的工作。
kafka是一个基于CA的系统(???)，zookeeper是一个基于CP的系统(很确定)，eureka是一个基于AP的系统(很确定)。

复制强一致
现有比较成熟的方案中，有两种保持强一致性复制的典型方法。这两种方法都需要副本中的一个被设计为leader，所有写入都需要发布到该副本。leader负责处理所有的接入。并广播这些写到其他follower副本，并且要保证复制顺序和leader的顺序一致。

第一种方法是基于法定人数。leader等待直到大多数副本收到数据。当leader出现故障，大多数follower会协调选举出新的leader。这种方法被用于Apache Zookeeper 和Google‘s Spanner.
第二种方法是leader等待所有副本收到数据(重要说明：在kafka中这个"所有"是所有的In-Sync Replicas)。如果leader出现故障，其他副本能被选举为新的leader。
kafka复制选择的是第二种方法，有两个主要原因：

相同数量的副本情况下，第二种方法能承受更多的容错。例如，总计2n+1个副本，第二种方法能承受2n个副本故障(只要还有一个ISR，就能正常写入)，而第一种方法只能承受n个副本故障。如果在只有两个副本的情况下，第一种方法不能容忍任意一个副本故障。

第一种方法延迟性表现更好，因为只需要法定人数确认即可，所以会隐藏掉一些比较慢的副本的影响。而kafka副本被设计在相同的数据中心的一个集群下。所以网络延迟这种变数是比较小的。

术语
为了了解kafka中的副本是如何实现的，我们首先需要介绍一些基本概念。在kafka中，消息流由topic定义，topic被切分为1个或者多个分区(partition)，复制发生在分区级别，每个分区有一个或者多个副本。

副本被均匀分配到kafka集群的不同服务器(称为broker)上。每个副本都维护磁盘上的日志。生产者发布的消息顺序追加到日志中，日志中每条消息被一个单调递增的offset标识。

offset是分区内的逻辑概念， 给定偏移量，可以在分区的每个副本中标识相同的消息。 当消费者订阅某个主题时，它会跟踪每个分区中的偏移量以供使用，并使用它来向broker发出获取消息的请求。

设计
kafka中增加副本的目标是为了更强的持久性和高可用。kafka要保证任何成功发布的消息不会丢失，且能被消费，即使在有一些服务器宕机的情况下。kafka复制的主要目标有：

可配置的持久化保证：例如，某些数据不能容忍丢失的应用，可以选择更强的持久性，当然会伴随延迟的增长。另一个产生海量允许部分数据丢失的应用，可以选择稍微弱一点的持久性，但是更获得更好的写入响应时间，得到更好的吞吐量。

自动化的副本管理：kafka要简化向broker分配副本的指配过程，并且能支持集群逐步扩容&缩容。

这样的话，有两个主要问题需要解决：

如何均匀的指配分区的副本给broker?
对于一个给定的分区，如何广播每条消息到其他副本?
数据复制
kafka允许客户端选择异步或者同步复制，异步复制的话，发布的消息，当被1个副本接收到就能确认。同步复制的话，kafka尽最大努力确保消息到达多个副本(所以有效的ISR)后才确认。当客户端尝试发布消息到一个topic的分区时，kafka必须传播这个消息到所有副本，kafka必须决定：

怎样传播;
在向客户端确认前，需要多少副本接收消息;
一个副本故障后，该怎么处理;
一个故障的副本恢复后该怎么处理;
实现
保持副本同步有两种常用的策略：主备复制和基于仲裁复制。这两种情况下，一个副本被设计为leader，其他副本被称为follower，所有写请求都由leader处理，leader传播写请求给follower。

在主备复制下，leader等待直到写在这个组里每个副本都完成，才向客户端发送确认。如果某个副本故障，leader把它从这个组移除，并继续写到剩余的副本。一个故障副本也被允许从新加入组，只要它恢复，并追赶上leader。在用n个副本的前提下，主备复制模式能容忍n-1个副本故障。

在基于仲裁方法下，leader等待直到写在大多数副本上完成，副本组的大小不会因为某些副本故障发生改变(例如某个分区有5个副本，即使有2个副本故障，我们还是认为这个副本组有5个副本)。因此如果有2n+1个副本，基于仲裁复制的话，只能容忍n个副本故障。如果leader出现故障，需要至少n+1个副本才能选举一个新的leader。

这两种方法需要权衡：

基于仲裁比主备有更好的写延迟，任何副本的延迟(例如FGC造成长时间的STW)将增加主备方法的写延迟，但是不会增加仲裁方法的写延迟。
在相同数量副本情况下，主备方法能容忍更多故障。
在主备方法前提下，副本因子是2，也能运行良好。但是在基于仲裁方法的复制，两个副本必须持续工作保持有效状态。
kafka选择主备复制，因为它能容忍更多副本故障，并且只有2个副本也能正常工作。
同步复制
kafka同步复制是典型的主备方式，每个分区有n个副本，并且能容忍n-1个副本故障。只有一个副本被选举为leader，其他都是follower。leader维护了一个ISR集合：这个副本集完全和leader保持同步状态，kafka还会把当前的leader和当前的ISR保持到zookeeper中。

每个副本保存信息在本地日志中，并且维护了一个日志中重要的offset位置。LEO表示日志尾部，HW是最新提交消息的offset。每个日志周期性的同步到磁盘，已经刷新的偏移量之前的数据保证保留在磁盘上。

写
为了发布消息到分区，客户端首先从zookeeper中找到分区的leader，然后发送消息到这个leader。leader写消息到它的本地日志，每个follower经常从leader拉取最新的消息。所以，follower接收到的所有消息的顺序和leader保持一致，follower把每条接收到的消息写入它的本地日志，并向leader发送一个确认。一旦leader接收到所有ISR副本的确认，消息就能被提交。leader推进HW，然后向客户端发送确认。为了更好的性能，每个follower在把消息写入内存后，就发送确认。因此，对于每条提交的消息，我们保证它被保存到多个副本的内容中然而，不保证任何副本已经持久化已提交消息到磁盘上。

由于这种相关故障相对罕见，并且这种方法能给我们一个在响应时间和持久性之间一个很好的平衡。在将来，kafka可能考虑增加一个选项参数从而提供更强的保证。

读
为了简化，读也是leader提供服务，并且只有HW以上的消息才会被暴露给消费者读取。

异步复制
为了支持异步复制，leader可以在消息写入本地日志后，马上通知客户端。唯一需要注意的是在追赶阶段，follower必须截断HW位置以后的数据。follower主要是异步复制，所以不能保证提交的消息在broker故障后不丢失。

复制实现
kafka复制示意图如下所示：

集群总计4个broker(broker1~broker4);
1个topic，2个分区，3个副本;
分区1即topic1-part1的leader在broker1上，分区2即topic1-part2的leader在broker4上;
producer写入消息到分区topic1-part1的leader上(在broker1上)，然后复制到它的两个副本，分别在broker2和broker3上。

producer写入消息到分区topic1-part2的leader上(在broker4上)，然后复制到它的两个副本，分别在broker2和broker3上。

当生产者发布消息到topic的某个分区时，消息首先被传递到leader副本，并追加日志。follower副本从leader中不停的拉取新消息，一旦有足够的副本收到消息，leader就会提交这个消息。

这里有个问题，leader是怎么决定什么是足够的。kafka维护了一个 in-sync replica(ISR)集合。这个ISR副本集都是存活的，并且完全赶上leader的副本，没有消息延迟(leader总是在ISR集合中)。当分区初始化创建时，每个副本都在ISR集合中。当新消息发布后，leader提交消息前一直等待直到所有ISR副本收到消息。如果某个follower副本故障，它将会被从ISR中移除。leader会继续提交新的消息，只不过ISR数量相比分区创建时副本数量更少。

请注意，现在，系统运行在under replicated模式。

leader还会维护high watermark (HW，可以翻译成高水位)，是指分区中最后一次提交消息的offset。HW会被不断传播给follower副本：

kafka high watermark
当一个故障副本被重启后，它首先从磁盘上恢复最新的HW，并将日志截断到HW。这是必要的，因为不能保证在HW之后的消息被提交，所以可能需要丢弃。然后副本成为follower，并继续从leader那里获取HW以后的消息。一旦完全赶上leader，这个副本从新被加入到ISR中。系统将重新回到fully replicated模式。

故障处理
kafka依赖zookeeper检测broker故障，kafka会用一个controller(broker集合中的一个)接收所有zookeeper关于故障，选举新leader等相关通知，这样还有一个好处，减少了对zookeeper的压力。如果某个leader故障，controller就会从ISR副本中选举一个新的leader，并发布新leader的消息给其他follower。

按照设计，leader选举过程中，已经提交的消息总是会被保留，一些未提交的消息可能会丢失。leader和每个分区的ISR也会被保存在Zookeeper中，controller出现故障转移时需要用到。由于broker级别的故障一般会非常少，所以预期的leader和ISR都会不经常改变。

对客户端来说，broker仅向消费者公开已经提交的消息。broker故障期间，已提交的数据始终被保留。消费者使用相同的offset可以从另一个被选举为leader的副本拉取消息。

生产者能选择在broker收到消息后何时得到broker的确认。例如，它能等到消息被leader提交并被所有ISR确认(即acks=-1)。另外，也可以选择消息只要被leader追加到日志中，可能还没有提交(acks=0表示无需等待leader确认，acks=1表示需要等待leader确认)。前一种情况即acks=-1，生产者需要等待更长的时间。但是确认的消息都保证在broker中保留。后一种情况即acks=0或者1，生产者有更低的延迟，更高的吞吐量，但一些确认的消息在broker故障时可能会丢失。如何抉择，由你决定。