Kafka大厂高频面试题:在保证高性能、高吞吐的同时保证高可用性

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: Kafka的消息传输保障机制非常直观。当producer向broker发送消息时,一旦这条消息被commit,由于副本机制(replication)的存在,它就不会丢失。但是如果producer发送数据给broker后,遇到的网络问题而造成通信中断,那producer就无法判断该条消息是否已经提交(commit)。虽然Kafka无法确定网络故障期间发生了什么,但是producer可以retry多次,确保消息已经正确传输到broker中,所以目前Kafka实现的是at least once。

Kafka的消息传输保障机制非常直观。当producerbroker发送消息时,一旦这条消息被commit,由于副本机制(replication)的存在,它就不会丢失。但是如果producer发送数据给broker后,遇到的网络问题而造成通信中断,那producer就无法判断该条消息是否已经提交(commit)。虽然Kafka无法确定网络故障期间发生了什么,但是producer可以retry多次,确保消息已经正确传输到broker中,所以目前Kafka实现的是at least once

一、幂等性

1.场景

所谓幂等性,就是对接口的多次调用所产生的结果和调用一次是一致的。生产者在进行重试的时候有可能会重复写入消息,而使用Kafka的幂等性功能就可以避免这种情况。

幂等性是有条件的:

  • 只能保证 Producer 在单个会话内不丢不重,如果 Producer 出现意外挂掉再重启是无法保证的(幂等性情况下,是无法获取之前的状态信息,因此是无法做到跨会话级别的不丢不重);
  • 幂等性不能跨多个 Topic-Partition,只能保证单个 partition 内的幂等性,当涉及多个Topic-Partition 时,这中间的状态并没有同步。

Producer 使用幂等性的示例非常简单,与正常情况下 Producer 使用相比变化不大,只需要把Producer 的配置 enable.idempotence 设置为 true 即可,如下所示:

Properties props = new Properties(); 
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); 
props.put("acks", "all"); // 当 enable.idempotence 为 true,这里默认为 all 
props.put("bootstrap.servers", "localhost:9092"); 
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
KafkaProducer producer = new KafkaProducer(props); 
producer.send(new ProducerRecord(topic, "test");

二、事务

1.场景

幂等性并不能跨多个分区运作,而事务可以弥补这个缺憾,事务可以保证对多个分区写入操作的原子性。操作的原子性是指多个操作要么全部成功,要么全部失败,不存在部分成功部分失败的可能。

为了实现事务,应用程序必须提供唯一的transactionalId,这个参数通过客户端程序来进行设定。

见代码库:
com.heima.kafka.chapter7.ProducerTransactionSend

properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);

2.前期准备

事务要求生产者开启幂等性特性,因此通过将transactional.id参数设置为非空从而开启事务特性的同时需要将
ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG
设置为true(默认值为true),如果显示设置为false,则会抛出异常。

KafkaProducer提供了5个与事务相关的方法,详细如下:

//初始化事务,前提是配置了transactionalId 
public void initTransactions() 
//开启事务 
public void beginTransaction() 
//为消费者提供事务内的位移提交操作 
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) 
//提交事务 
public void commitTransaction() 
//终止事务,类似于回滚 
public void abortTransaction()

3.案例解析

见代码库:
com.heima.kafka.chapter7.ProducerTransactionSend

消息发送端

/**
    * Kafka Producer事务的使用 
    */ 
public class ProducerTransactionSend { 
    public static final String topic = "topic-transaction"; 
    public static final String brokerList = "localhost:9092"; 
    public static final String transactionId = "transactionId"; 
    public static void main(String[] args) { 
        Properties properties = new Properties(); 
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList); 
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId); 
        KafkaProducer<String, String> producer = new KafkaProducer<> (properties); 
        producer.initTransactions(); 
        producer.beginTransaction(); 
        try {
            //处理业务逻辑并创建ProducerRecord 
            ProducerRecord<String, String> record1 = new ProducerRecord<>(topic, "msg1"); 
            producer.send(record1); 
            ProducerRecord<String, String> record2 = new ProducerRecord<>(topic, "msg2"); 
            producer.send(record2); 
            ProducerRecord<String, String> record3 = new ProducerRecord<>(topic, "msg3"); 
            producer.send(record3); 
            //处理一些其它逻辑 
            producer.commitTransaction(); 
        } catch (ProducerFencedException e) { 
            producer.abortTransaction(); 
        }
        producer.close(); 
    } 
}

模拟事务回滚案例

try { 
    //处理业务逻辑并创建ProducerRecord 
    ProducerRecord<String, String> record1 = new ProducerRecord<>(topic, "msg1"); 
    producer.send(record1); 
    //模拟事务回滚案例 
    System.out.println(1/0); 
    ProducerRecord<String, String> record2 = new ProducerRecord<>(topic, "msg2"); 
    producer.send(record2); 
    ProducerRecord<String, String> record3 = new ProducerRecord<>(topic, "msg3"); 
    producer.send(record3); 
    //处理一些其它逻辑 
    producer.commitTransaction(); 
} catch (ProducerFencedException e) { 
    producer.abortTransaction(); 
}

从上面案例中,msg1发送成功之后,出现了异常事务进行了回滚,则msg1消费端也收不到消息。

三、控制器

在Kafka集群中会有一个或者多个broker,其中有一个broker会被选举为控制器(Kafka Controller),它负责管理整个集群中所有分区和副本的状态。当某个分区的leader副本出现故障时,由控制器负责为该分区选举新的leader副本。当检测到某个分区的ISR集合发生变化时,由控制器负责通知所有broker更新其元数据信息。当使用kafka-topics.sh脚本为某个topic增加分区数量时,同样还是由控制器负责分区的重新分配。

Kafka中的控制器选举的工作依赖于Zookeeper,成功竞选为控制器的broker会在Zookeeper中创建/controller这个临时(EPHEMERAL)节点,此临时节点的内容参考如下:

1.ZooInspector管理

  • 使用zookeeper图形化的客户端工具(ZooInspector)提供的jar来进行管理,启动如下:
  1. 定位到jar所在目录
  2. 运行jar文件 java -jar zookeeper-dev-ZooInspector.jar
  3. 连接Zookeeper

网络异常,图片无法展示
|

{"version":1,"brokerid":0,"timestamp":"1529210278988"}

网络异常,图片无法展示
|


其中version在目前版本中固定为1,brokerid表示
称为控制器的broker的id编号,timestamp表示竞选称为控制器时的时间戳

在任意时刻,集群中有且仅有一个控制器。每个broker启动的时候会去尝试去读取/controller节点的brokerid的值,如果读取到brokerid的值不为-1,则表示已经有其它broker节点成功竞选为控制器,所以当前broker就会放弃竞选;如果Zookeeper中不存在/controller这个节点,或者这个节点中的数据异常,那么就会尝试去创建/controller这个节点,当前broker去创建节点的时候,也有可能其他broker同时去尝试创建这个节点,只有创建成功的那个broker才会成为控制器,而创建失败的broker则表示竞选失败。每个broker都会在内存中保存当前控制器的brokerid值,这个值可以标识为activeControllerId

Zookeeper中还有一个与控制器有关的/controller_epoch节点,这个节点是持久(PERSISTENT)节点,节点中存放的是一个整型的controller_epoch值。controller_epoch用于记录控制器发生变更的次数,即记录当前的控制器是第几代控制器,我们也可以称之为“控制器的纪元”

网络异常,图片无法展示
|


controller_epoch的初始值为1,即集群中第一个控制器的纪元为1,当控制器发生变更时,没选出一个新的控制器就将该字段值加1。每个和控制器交互的请求都会携带上controller_epoch这个字段,如果请求的controller_epoch值小于内存中的controller_epoch值,则认为这个请求是向已经过期的控制器所发送的请求,那么这个请求会被认定为无效的请求。如果请求的controller_epoch值大于内存中的controller_epoch值,那么则说明已经有新的控制器当选了。由此可见,Kafka通过controller_epoch来保证控制器的唯一性,进而保证相关操作的一致性。

具备控制器身份的broker需要比其他普通的broker多一份职责,具体细节如下:

  1. 监听partition相关的变化。
  2. 监听topic相关的变化。
  3. 监听broker相关的变化。
  4. 从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。

四、可靠性保证

  1. 可靠性保证:确保系统在各种不同的环境下能够发生一致的行为
  2. Kafka的保证
  3. 保证分区消息的顺序如果使用同一个生产者同一个分区写入消息,而且消息B在消息A之后写入那么Kafka可以保证消息B的偏移量比消息A的偏移量大,而且消费者会先读取消息A再读取消息B
  4. 只有当消息被写入分区的所有同步副本时(文件系统缓存),它才被认为是已提交
  5. 生产者可以选择接收不同类型的确认,控制参数 acks
  6. 只要还有一个副本是活跃的,那么已提交的消息就不会丢失
  7. 消费者只能读取已经提交的消息

1. 失效副本

怎么样判定一个分区是否有副本是处于同步失效状态的呢?从Kafka 0.9.x版本开始通过唯一的一个参数replica.lag.time.max.ms(默认大小为10,000)来控制,当ISR中的一个follower副本滞后leader副本的时间超过参数replica.lag.time.max.ms指定的值时即判定为副本失效,需要将此follower副本剔出除ISR之外。具体实现原理很简单,当follower副本将leader副本的LEO(Log End Offset,每个分区最后一条消息的位置)之前的日志全部同步时,则认为该follower副本已经追赶上leader副本,此时更新该副本的lastCaughtUpTimeMs标识。Kafka的副本管理器(ReplicaManager)启动时会启动一个副本过期检测的定时任务,而这个定时任务会定时检查当前时间与副本的lastCaughtUpTimeMs差值是否大于参数replica.lag.time.max.ms指定的值。千万不要错误地认为follower副本只要拉取leader副本的数据就会更新lastCaughtUpTimeMs,试想当leader副本的消息流入速度大于follower副本的拉取速度时,follower副本一直不断的拉取leader副本的消息也不能与leader副本同步,如果还将此follower副本置于ISR中,那么当leader副本失效,而选取此follower副本为新的leader副本,那么就会有严重的消息丢失。

2.副本复制

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

一个副本可以不同步Leader有如下几个原因 慢副本:在一定周期时间内follower不能追赶上leader。最常见的原因之一是I / O瓶颈导致follower追加复制消息速度慢于从leader拉取速度。 卡住副本:在一定周期时间内follower停止从leader拉取请求。follower replica卡住了是由于GC暂停follower失效或死亡

新启动副本:当用户给主题增加副本因子时,新的follower不在同步副本列表中,直到他们完全赶上了leader日志。

如何确定副本是滞后的

replica.lag.max.messages=4

网络异常,图片无法展示
|


在服务端现在只有一个参数需要配置replica.lag.time.max.ms。这个参数解释replicas响应partition leader的最长等待时间。检测卡住或失败副本的探测——如果一个replica失败导致发送拉取请求时间间隔超过replica.lag.time.max.ms。Kafka会认为此replica已经死亡会从同步副本列表从移除。检测慢副本机制发生了变化——如果一个replica开始落后leader超过replica.lag.time.max.ms。Kafka会认为太缓慢并且会从同步副本列表中移除。除非replica请求leader时间间隔大于replica.lag.time.max.ms,因此即使leader使流量激增和大批量写消息。Kafka也不会从同步副本列表从移除该副本。

1.Leader Epoch引用

数据丢失场景

网络异常,图片无法展示
|

数据出现不一致场景

网络异常,图片无法展示
|

2.Kafka 0.11.0.0.版本解决方案

造成上述两个问题的根本原因在于HW值被用于衡量副本备份的成功与否以及在出现failture时作为日志截断的依据,但HW值得更新是异步延迟的,特别是需要额外的FETCH请求处理流程才能更新,故这中间发生的任何崩溃都可能导致HW值的过期。鉴于这些原因,Kafka 0.11引入了leader epoch来取代HW值。Leader端多开辟一段内存区域专门保存leader的epoch信息,这样即使出现上面的两个场景也能很好地规避这些问题。

所谓leader epoch实际上是一对值:(epochoffset)。epoch表示leader的版本号,从0开始,当leader变更过1次时epoch就会+1,而offset则对应于该epoch版本的leader写入第一条消息的位移。因此假设有两对值:

  1. (0, 0)
  2. (1, 120)

则表示第一个leader从位移0开始写入消息;共写了120条[0, 119];而第二个leader版本号是1,从位移120处开始写入消息。

leader broker中会保存这样的一个缓存,并定期地写入到一个checkpoint文件中。

避免数据丢失:

网络异常,图片无法展示
|

避免数据不一致

网络异常,图片无法展示
|

六、消息重复的场景及解决方案

1.生产者端重复

生产发送的消息没有收到正确的broke响应,导致producer重试。

producer发出一条消息,broke落盘以后因为网络等种种原因发送端得到一个发送失败的响应或者网络中断,然后producer收到一个可恢复的Exception重试消息导致消息重复。

解决方案:

  1. 启动kafka的幂等性

要启动kafka的幂等性,无需修改代码,默认为关闭,需要修改配置文件:enable.idempotence=true 同时要求 ack=allretries>1

  1. ack=0,不重试。

可能会丢消息,适用于吞吐量指标重要性高于数据丢失,例如:日志收集。

2.消费者端重复

  1. 根本原因

数据消费完没有及时提交offset到broker。

解决方案

  1. 取消自动自动提交

每次消费完或者程序退出时手动提交。这可能也没法保证一条重复。

  1. 下游做幂等

一般的解决方案是让下游做幂等或者尽量每消费一条消息都记录offset,对于少数严格的场景可能需要把offset或唯一ID,例如订单ID和下游状态更新放在同一个数据库里面做事务来保证精确的一次更新或者在下游数据表里面同时记录消费offset,然后更新下游数据的时候用消费位点做乐观锁拒绝掉旧位点的数据更新。

七、__consumer_offsets

_consumer_offsets是一个内部topic,对用户而言是透明的,除了它的数据文件以及偶尔在日志中出现这两点之外,用户一般是感觉不到这个topic的。不过我们的确知道它保存的是Kafka新版本consumer的位移信息。

1.何时创建

一般情况下,当集群中第一有消费者消费消息时会自动创建主题__consumer_offsets,分区数可以通过
offsets.topic.num.partitions参数
设定,默认值为50,如下

网络异常,图片无法展示
|

2.解析分区

见代码库:
com.heima.kafka.chapter7.ConsumerOffsetsAnalysis

获取所有分区

网络异常,图片无法展示
|

总结

本章主要讲解了Kafka相关稳定性的操作,包括幂等性、事务的处理,同时对可靠性保证与一致性保证做了讲解,讲解了消息重复以及解决方案。

本文就是愿天堂没有BUG给大家分享的内容,大家有收获的话可以分享下,想学习更多的话可以到微信公众号里找我,我等你哦。

相关文章
|
9天前
|
消息中间件 分布式计算 监控
Python面试:消息队列(RabbitMQ、Kafka)基础知识与应用
【4月更文挑战第18天】本文探讨了Python面试中RabbitMQ与Kafka的常见问题和易错点,包括两者的基础概念、特性对比、Python客户端使用、消息队列应用场景及消息可靠性保证。重点讲解了消息丢失与重复的避免策略,并提供了实战代码示例,帮助读者提升在分布式系统中使用消息队列的能力。
43 2
|
9天前
|
消息中间件 监控 大数据
Kafka消息队列架构与应用场景探讨:面试经验与必备知识点解析
【4月更文挑战第9天】本文详尽探讨了Kafka的消息队列架构,包括Broker、Producer、Consumer、Topic和Partition等核心概念,以及消息生产和消费流程。此外,还介绍了Kafka在微服务、实时数据处理、数据管道和数据仓库等场景的应用。针对面试,文章解析了Kafka与传统消息队列的区别、实际项目挑战及解决方案,并展望了Kafka的未来发展趋势。附带Java Producer和Consumer的代码示例,帮助读者巩固技术理解,为面试做好准备。
34 0
|
1天前
|
消息中间件 监控 Java
腾讯面试:如何提升Kafka吞吐量?
Kafka 是一个分布式流处理平台和消息系统,用于构建实时数据管道和流应用。它最初由 LinkedIn 开发,后来成为 Apache 软件基金会的顶级项目。 Kafka 特点是**高吞吐量、分布式架构、支持持久化、集群水平扩展和消费组消息消费**,具体来说: 1. **高吞吐量**:Kafka 具有高性能和低延迟的特性,能够处理大规模数据,并支持每秒数百万条消息的高吞吐量。 2. **分布式架构**:Kafka 采用分布式架构,可以水平扩展,多个节点之间能够实现负载均衡和高可用性。 3. **可持久化**:Kafka 将消息持久化到磁盘中,保证消息的可靠性,即使消费者下线或出现故障,消
|
7天前
|
消息中间件 关系型数据库 MySQL
MySQL 到 Kafka 实时数据同步实操分享(1),字节面试官职级
MySQL 到 Kafka 实时数据同步实操分享(1),字节面试官职级
|
8天前
|
消息中间件 Java Kafka
Java大文件排序(有手就能学会),kafka面试题2024
Java大文件排序(有手就能学会),kafka面试题2024
|
8天前
|
消息中间件 前端开发 Java
java面试刷题软件kafka和mq的区别面试
java面试刷题软件kafka和mq的区别面试
|
9天前
|
消息中间件 Kafka API
这些年背过的面试题——Kafka篇
本文是技术人面试系列Kafka篇,面试中关于Kafka都需要了解哪些基础?一文带你详细了解,欢迎收藏!
|
9天前
|
消息中间件 网络协议 Dubbo
Kafka常见面试题
Kafka常见面试题
33 2
|
9天前
|
消息中间件 存储 监控
深入剖析:Kafka流数据处理引擎的核心面试问题解析75问(5.7万字参考答案)
Kafka 是一款开源的分布式流处理平台,被广泛应用于构建实时数据管道、日志聚合、事件驱动的架构等场景。本文将深入探究 Kafka 的基本原理、特点以及其在实际应用中的价值和作用。 Kafka 的基本原理是建立在发布-订阅模式之上的。生产者将消息发布到主题(Topic)中,而消费者则可以订阅这些主题并处理其中的消息。Kafka包括多个关键组件,如生产者、消费者、主题分区、ZooKeeper 等,Kafka 实现了高性能的消息传递和存储。特点:高吞吐量、可持久化存储、水平扩展、容错性和实时性等。
119 0
|
7天前
|
消息中间件 关系型数据库 Kafka
实时计算 Flink版产品使用合集之想要加快消费 Kafka 数据的速度,该怎么配置参数
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
20 2

热门文章

最新文章