更多相关内容可查看

kafka高可用设计

Apache Kafka 的高可用设计主要通过以下几个方面来实现：

• 副本机制（Replication）：在 Kafka 中，每个 partition 都有多个副本，其中一个作为 leader，其他的作为follower。所有的读写操作都是通过 leader 来进行的，follower 则负责从 leader 同步数据。当 leader出现故障时，会从 follower 中选举出新的 leader，以保证服务的可用性。
• 分区机制（Partitioning）：Kafka 的 topic 可以分为多个 partition，每个 partition可以在不同的服务器上，这样即使某个服务器出现故障，也不会影响到其他 partition 的正常服务。
• 消费者组（Consumer Groups）：Kafka 允许多个消费者组同时消费同一个 topic，每个消费者组都会维护自己的offset，这样即使某个消费者组出现故障，也不会影响到其他消费者组的消费
• ZooKeeper 集群：Kafka 使用 ZooKeeper 来管理集群的元数据信息，如 broker、topic 和
  partition 的信息等。ZooKeeper 本身也是一个分布式服务，可以通过多个节点组成集群，提供高可用性。
• ISR（In-Sync Replicas）机制：Kafka 通过 ISR 机制来保证数据的一致性。只有在 ISR 列表中的
  follower 才有资格被选为新的 leader，这样可以保证新的 leader 拥有所有的数据副本。

通过以上的设计，Kafka 能够在面对故障时，仍能保证服务的可用性和数据的一致性。

集群架构

Kafka 的服务器端由被称为 Broker 的服务进程构成，即一个 Kafka 集群由多个 Broker 组成
这样如果集群中某一台机器宕机，其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的基础

Kafka集群选举

ISR与OSR

Kafka为了对消息进行分类，引入了Topic（主题）的概念。生产者在发送消息的时候，需要指定发送到某个Topic，然后消息者订阅这个Topic并进行消费消息。

Kafka为了提升性能，又在Topic的基础上，引入了Partition（分区）的概念。Topic是逻辑概念，而Partition是物理分组。一个Topic可以包含多个Partition，生产者在发送消息的时候，需要指定发送到某个Topic的某个Partition，然后消息者订阅这个Topic并消费这个Partition中的消息。

Kafka为了提高系统的吞吐量和可扩展性，把一个Topic的不同Partition放到多个Broker节点上，充分利用机器资源，也便于扩展Partition。

Kafka为了保证数据的安全性和服务的高可用，又在Partition的基础上，引入Replica（副本）的概念。一个Partition包含多个Replica，Replica之间是一主多从的关系，有两种类型Leader Replica（领导者副本）和Follower Replica（跟随者副本），Replica分布在不同的Broker节点上。

Leader Replica负责读写请求，Follower Replica只负责同步Leader Replica数据，不对外提供服务。当Leader Replica发生故障，就从Follower Replica选举出一个新的Leader Replica继续对外提供服务，实现了故障自动转移。

Kafka为了提升Replica的同步效率和数据写入效率，又对Replica进行分类。针对一个Partition的所有Replica集合统称为AR（Assigned Replicas，已分配的副本），包含Leader Replica和Follower Replica。与Leader Replica保持同步的Replica集合称为ISR（In-Sync Replicas，同步副本），与Leader Replica保持失去同步的Replica集合称为OSR（Out-of-Sync Replicas，失去同步的副本），AR = ISR + OSR。
Leader Replica将消息写入磁盘前，需要等ISR中的所有副本同步完成。如果ISR中某个Follower Replica同步数据落后Leader Replica过多，会被转移到OSR中。如果OSR中的某个Follower Replica同步数据追上了Leader Replica，会被转移到ISR中。当Leader Replica发生故障的时候，只会从ISR中选举出新的Leader Replica

LEO和HW

Kafka为了记录副本的同步状态，以及控制消费者消费消息的范围，于是引入了LEO（Log End Offset，日志结束偏移量）和HW（High Watermark，高水位）。
LEO表示分区中的下一个被写入消息的偏移量，也是分区中的最大偏移量。LEO用于记录Leader Replica和Follower Replica之间的数据同步进度，每个副本中各有一份。

Leader : LEO 1000 HW : 950
Follower1 : LEO 980
Follower2 : LEO 1000
Follower3 : LEO 950

HW表示所有副本（Leader和Follower）都已成功复制的最小偏移量，是所有副本共享的数据值。换句话说，HW之前的消息都被视为已提交，消费者可以消费这些消息。用于确保消息的一致性和只读一次。

LEO和HW的更新流程：

1. 初始状态，三个副本中各有0和1两条消息，LEO都是2，位置2是空的，表示是即将被写入消息的位置。HW也都是2，表示Leader Replica中的所有消息已经全部同步到Follower Replica中，消费者可以消费0和1两条消息。

2. 生产者往Leader Replica中发送两条消息，此时Leader Replica的LEO的值增加2，变成4。由于还没有开始往Follower Replica同步消息，所以HW值和Follower Replica中LEO值都没有变。由于消费者只能消费HW之前的消息，也就是0和1两条消息

3. Leader Replica开始向Follower Replica同步消息，同步速率不同，Follower1的两条消息2和3已经同步完成，而Follower2只同步了一条消息2。此时，Leader和Follower1的LEO都是4，而Follower2的LEO是3，HW表示已成功同步的最小偏移量，值是3，表示此时消费者只能读到0、1、2，三条消息

4. 所有消息都同步完成，三个副本的LEO都是4，HW也是4，消费者可以读到0、1、2、3，四条消息

Kafka分区Leader选举

常见的有以下几种情况会触发Partition的Leader Replica选举：

1. Leader Replica 失效：当 Leader Replica 出现故障或者失去连接时，Kafka 会触发 Leader Replica 选举。
2. Broker 宕机：当 Leader Replica 所在的 Broker 节点发生故障或者宕机时，Kafka 也会触发 Leader Replica 选举。
3. 新增 Broker：当集群中新增 Broker 节点时，Kafka 还会触发 Leader Replica 选举，以重新分配 Partition 的 Leader。
4. 新建分区：当一个新的分区被创建时，需要选举一个 Leader Replica。
5. ISR 列表数量减少：当 Partition 的 ISR 列表数量减少时，可能会触发 Leader Replica 选举。当 ISR 列表中副本数量小于Replication Factor（副本因子）时，为了保证数据的安全性，就会触发 Leader Replica 选举。
6. 手动触发：通过 Kafka 管理工具（kafka-preferred-replica-election.sh），可以手动触发选举，以平衡负载或实现集群维护

Leader Replica选举策略

在 Kafka 集群中，常见的 Leader Replica 选举策略有以下三种：

1. ISR 选举策略：默认情况下，Kafka 只会从 ISR 集合的副本中选举出新的 Leader Replica，OSR 集合中的副本不具备参选资格。
2. 不干净副本选举策略（Unclean Leader Election）：在某些情况下，ISR 选举策略可能会失败，例如当所有 ISR 副本都不可用时。在这种情况下，可以使用 Unclean Leader 选举策略。Unclean Leader 选举策略会从所有副本中（包含OSR集合）选择一个副本作为新的 Leader 副本，即使这个副本与当前 Leader 副本不同步。这种选举策略可能会导致数据丢失，默认关闭
3. 首选副本选举策略（Preferred Replica Election）：首选副本选举策略也是 Kafka 默认的选举策略。在这种策略下，每个分区都有一个首选副本（Preferred Replica），通常是副本集合中的第一个副本。当触发选举时，控制器会优先选择该首选副本作为新的 Leader Replica，只有在首选副本不可用的情况下，才会考虑其他副本。
   当然，可以使用命令手动指定每个分区的首选副本：

bin/kafka-topics.sh --zookeeper localhost:2181 --topic my-topic-name --replica-assignment 0:1,1:2,2:0 --partitions 3
意思是：my-topic-name有3个partition，partition0的首选副本是Broker1，partition1首选副本是Broker2，partition2的首选副本是Broker0

Leader Replica选举过程

谁来主持选举？
kafka先在brokers里面选一个broker作为Controller主持选举。Controller是使用zookeeper选举出来的，每个broker都往zk里面写一个/controller节点，谁先写成功，谁就成为Controller。如果Controller失去连接，zk上的临时节点就会消失。其它的broker通过watcher监听到Controller下线的消息后，开始选举新的Controller。

一个Broker节点相当于一台机器，多个Broker节点组成一个Kafka集群。Controller节点也叫控制器节点 , 他负责直接与zookeeper进行通信，并负责管理整个集群的状态和元数据信息

Controller的责任

• 监听Broker的变化
• 监听Topic变化
• 监听Partition变化
• 获取和管理Broker、Topic、Partition的信息
• 管理Partition的主从信息

当Leader Replica宕机或失效时，就会触发 Leader Replica 选举，分为两个阶段，第一个阶段是候选人的提名和投票阶段，第二个阶段是Leader的确认阶段。具体过程如下：

lag(滞后）是kafka消费队列性能监控的重要指标，lag的值越大，表示kafka的消息堆积越严重

1. 候选人提名和投票阶段
   在Leader Replica失效时，ISR集合中所有Follower Replica都可以成为新的Leader Replica候选人。每个Follower Replica会在选举开始时向其他Follower Replica发送成为候选人的请求，并附带自己的元数据信息，包括自己的当前状态和Lag值。而Preferred replica优先成为候选人。
   其他Follower Replica在收到候选人请求后，会根据请求中的元数据信息，计算每个候选人的Lag值，并将自己的选票投给Lag最小的候选人。如果多个候选人的Lag值相同，则随机选择一个候选人。
2. Leader确认阶段
   在第一阶段结束后，所有的Follower Replica会重新计算每位候选人的Lag值，并投票给Lag值最小的候选人。此时，选举的结果并不一定出现对候选人的全局共识。为了避免出现这种情况，Kafka中使用了ZooKeeper来实现分布式锁，确保只有一个候选人能够成为新的Leader Replica。
   当ZooKeeper确认有一个候选人已经获得了分布式锁时，该候选人就成为了新的Leader Replica，并向所有的Follower Replica发送一个LeaderAndIsrRequest请求，更新Partition的元数据信息。其他Follower Replica接收到请求后，会更新自己的Partition元数据信息，将新的Leader Replica的ID添加到ISR列表中

副本机制(Replication）

Kafka 中消息的备份又叫做 副本（Replica）
Kafka 定义了两类副本：

• 领导者副本（Leader Replica）: 负责数据读写
• 追随者副本（Follower Replica）: 只负责数据备份
• 当领导者副本所在节点宕机之后, 会从追随者副本中选举一个节点, 升级为领导者副本 , 对外提供数据读写服务, 保证数据安全

消费者组和再均衡

消费者组

消费者组（Consumer Group）是由一个或多个消费者实例（Consumer Instance）组成的群组，具有可扩展性和可容错性的一种机制。消费者组内的消费者共享一个消费者组ID，这个ID 也叫做 Group ID，组内的消费者共同对一个主题进行订阅和消费，同一个组中只能够由一个消费者去消费某一个分区的数据，多余的消费者会闲置，派不上用场。

同一个分区只能被一个消费者组中的一个消费者消费 , 一个消费者组中的某一个消费者, 可以消费多个分区

一个生产者发送一条消息只能被一个消费者消费 : 让消费者处于同一个组中即可
一个生产者发送一条消息需要被多个消费者消费 : 让消费者处于不同的组中

@Component
public class KafkaConsumerListener {

    @KafkaListener(topics = "kafka.topic.my-topic1",groupId = "group1")
    public void listenTopic1group1(ConsumerRecord<String, String> record) {
        String key = record.key();
        String value = record.value();
        System.out.println("group1中的消费者接收到消息:"+key + " : " + value);
    }

    @KafkaListener(topics = "kafka.topic.my-topic1",groupId = "group2")
    public void listenTopic1group2(ConsumerRecord<String, String> record) {
        String key = record.key();
        String value = record.value();
        System.out.println("group2中的消费者接收到消息:"+key + " : " + value);
    }
}

再均衡(重平衡)

再均衡就是指 当消费者组中的消费者发生变更的时候(新增消费者, 消费者宕机) , 重新为消费者分配消费分区的过程

当消费者组中重新加入消费者 , 或者消费者组中有消费者宕机 , 这个时候Kafka会为消费者组中的消费者从新分配消费分区的过程就是再均衡

重平衡(再均衡)非常重要，它为消费者群组带来了高可用性 和 伸缩性，我们可以放心的添加消费者或移除消费者，不过在正常情况下我们并不希望发生这样的行为。在重平衡期间，消费者无法读取消息，造成整个消费者组在重平衡的期间都不可用 , 并且在发生再均衡的时候有可能导致消息的丢失和重复消费