刚刚学了kafka，整理一些面试题知识点，帮助记忆

本文是自己学习整理，以及参考其他博客，如有什么错误，欢迎指出，谢谢~~
kafka参数配置官方文档
其他博主博客

1、什么是kafka

• 传统定义： Kafka是一个分布式的基于发布/订阅模式的消息队列（MQ）,主要应用于大数据实时处理领域
• 最新定义： Kafka是一个开源平台的分布式事件流平台（Event Streaming Platform），大部分公司应用于数据管道、流分析、数据集成和关键任务应用
• 注：相关概念
  • 注册中心zookeeper（非kafka内部），2.8.0后kafka版本抛弃了zookeeper
  • Producer： 消息生产者，向kafka集群写入数据
  • Consumer： 消息消费者，获取kafka集群数据
  • Consumer Group： 消费者组，由多个consumer组成；消费者组内每个消费者消费不同分区数据，1个分区只能由1个组内的1个消费者消费；消费者之间互不影响；消费者组是逻辑上的一个订阅者
  • Broker： kafka一台服务器就是1个Broker，1个集群由多个broker组成；1个borker中容纳多个topic
  • Topic： 消息主题，将消息进行分类，可以理解为1个队列
  • Partition： 为了实现扩展性，1个topic可以分为多个partition，每个partition都是一个有序队列
  • Replica： 副本，1个topic的每个分区都有若干副本，一个leader和若干follower
  • Leader： 多个分区中有1个为主副本，即leader，生产者和消费者交流数据的对象，其余均为follower
  • Follower： 多个分区除了Leader外的其余副本，会实时从leader中同步数据，保持于leader数据的同步，Leader发生故障市，Follower会选举成为新的Leader，leader不对外服务
  • Coordinator： 协调者，为了消费者组分配分区以及冲平衡Rebalance操作

2、什么是消息队列

目前主流消息队列有Kafka、ActiveMQ、RabbitMQ、RocketMQ; 大数据场景多采用kafka,JavaEE多采用其余三种

1）、应用场景

• 缓存/削峰： 控制和优化数据流经过系统的速度，解决生产者和消费者消息处理 不一致问题；缓解流量洪峰时服务器压力
• 解耦： 通过遵循同样的接口约束，允许两边不同系统独立地扩展和修改内部逻辑
• 异步通信： 允许用户将一个消息放入队列，并不立即进行处理，在需要地时候再去处理

2）、模式

• 点对点模式： 消费者主动拉取数据，收到后清除消息
• 发布/订阅模式：
  • 可以有多个topic主题
  • 消费者消费数据后，不删除数据
  • 每个消费者相互独立，都可以消费到数据

3）、kafka通过Consumer Group同时支持两种模式，所有的消费者在一个组内，消息只能被同一个Group内的一个消费者消费，就是点对点模式；如果组内仅有一个消费者，就是发布/订阅模式

3、kafka通信流程

• 1、kafka集群启动后，通过server.properties配置参数自动注册到zookeeper，再分目录下创建kafka目录，同时订阅zookeeper的brokers/ids路径，保存所有的broker信息
• 生产者启动通过bootstrap.servers连接到指定的broker上（代码开发过程中bootstrap.servers要使用域名，ip可能连接不上），创建tcp连接
• 和所有的broker创建连接
• 然后生产者者开始发送数据到kafak集群
• 消费者和生产者一样连接上broker
• 获取到broker的元数据，根据分区Leader节点所在的broker节点，和broker创建连接
• 开始消费数据

4、Leader选举流程

• 前提： kafka集群中有1个broker的Controller会被选举为Controller Leader，负责管理集群broker的上下线、所有topic的分区副本和Leader选举工作，Controller的信息同步工作依赖zookeeper
• 流程： 当不存在Leader或者Leader挂掉了，会按照在ISR中存活为前提，在AR中排在前面的优先的规则进行选取Leader
• eg： ar:[1,0,2], isr:[1,0,2],那么Leader会按照1，0，2的顺序进行轮询
• 注：2.8.0以前，Leader信息记录在zk，之后采用kraft版本就不再需要zk，记录在kafka内部

5、副本及同步原理

副本同步是Follower从Leader同步数据，作为冗余灾备使用，不对外，默认1个，生产上基本>=2个，Leader负责对外提供服务。

• AR： Assigned Replicas，所有副本集合
• ISR： 和Leader保持同步的副本，默认同步时间（replica.lag.time.max.ms）默认10s，Leader和Follower时差10s内就可以认为是同步
• LEO（Log End Offset）： 下一条待写入消息的位移（见下图）
• HW（High Watermark）： 高水位（复制点），所有副本中最小的LEO（见下图）
  
• 同步过程：
  • 1、当Leader中没有数据可同步，则Follower阻塞；Leader收到消费者的数据后解除Follower阻塞，Follower开始同步数据；每次发送fetch请求拉取数据。
  • 2、HW计算：当前broker1同步到5，broker2同步到7，HW=min(LEO_Broder0，LEO_Broder1， LEO_Broder2)= 5;
• Leader故障处理：
  • Leader故障后会从ISR中选出一个新的Leader
  • 为保证多个副本之间数据的一致性，其余的Follower会先将各自log中高于HW的部分截掉，然后从新的Leader同步数据
  • 注：这只能保证副本之间的数据一致性，不保证数据不丢失或者不重复
• Follower故障处理：
  • 故障的Follower会临时踢出ISR
  • 期间Leader和其他Follower继续接收同步数据
  • 待该Follower恢复后，会读取本地磁盘记录的上次HW，并将log文件高于HW的部分截取掉，从HW开始向Leader进行同步数据
  • 等该LEO_Follower>=HW_Leader，既追上了Leader，就可以重新加入ISR

6、消费者消费数据的方式

kafka中consumer采用的是poll（拉取）模式，没有采用pull（推送模式）,是因为由broker决定消息发送速率很难适应所有的消费者消费速率；poll模式的不足之处在于如果broker中没有数据，消费者会陷入空循环

7、分区分配策略以及原理

• 4种策略：Range，RoundRobin，Sticky，CooperativeSticky(3.0以后版本)

• 默认策略： Range + CooperativeSticky，可以通过partition.assignment.strategy参数进行调整，同时可以使用多种策略

• 再平衡：

  • 每个消费者都会和coordinator保持心跳（默认3s），一旦超时（session.timeout.ms=45s），该消费者就会被移除，触发再平衡；或者消费者处理消息时间过长（max.poll.interval.ms=5mins），也会触发再平衡
  • 一般极力避免再平衡，再平衡过程中会停止对数据的消费，造成数据积压

• Range：
  
  如上图所示，range分区尽可能保证分区均匀，通过Count_partitions/Count_consumer来决定每个消费者应该消费几个分区，如果除不尽，排在前面的消费者就会多消费分区

  • 注： 如果只有一个topic，按照上面的场景consumer-0多消费一个没问题，如果是N个topic，并且每一个topic分区和上面一样，那么consumer-0就会多消费N个partition，容易产生数据倾斜

• RoundRobin：
  

  • RoundRobin针对集群种所有的partition和consumer都列出来，然后按照hashcode进行排序，最后通过轮询算法分配partition到各个消费者

• Sitcky：

  • 顾名思义，粘性分区就是再进行新一次分配前，考虑上一次的分配结果，尽量少的调整分配的变动，可以节省大量的开销
  • 自kafak 0.11.x版本引入，首先会尽量均匀地方式分区到消费者上，再出现同一组消费者组内消费者出现问题的时候，会尽量保持原有分配地分区不变化

• CooperativeSticky：

  • 合作者粘性保持Sticky的逻辑之外，同时允许合作者再平衡

8、如何保证消息的可靠性

• 生产者

  • acks应答级别
    • 0： 生产者发送数据到集群后就结束，可靠性差，效率高；生产上基本不使用
    • 1： 生产者发送数据到集群后Leader应答，可靠性中等，效率中等，多用在日志的传输上，允许一定的丢失
    • -1(all)： 生产者发送数据到集群后，Leader和ISR队列里面所有的Follower应答，可靠性高，效率低，多用在关键数据、对可靠性要求比较高的场景
  • 消息失败重试
    • retries： 此参数让生产者发送消息失败后不停重试，默认是INT_MAX，如果设置了重试还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTRS_PER_CONNECTION=1，否则再重试此失败消息的时候，可能有其他消息发送成功了

• 消费者

  • 关闭自动提交位移，修改业务处理成功手动提交即可使得消息丢失
  • 消费者再平衡的时候，会读取上次提交的偏移量，默认是5s，会导致重复消费或者丢失消息
  • enable.auto.commit=false设置为手动提交

• 注：数据完全可靠条件： ACK级别设置为-1 + 分区副本>=2 + ISR里应答的最小副本数量>=2

9、数据有序/乱序

• 生产者发送过来的数据做到单分区内有序，多分区无序
• 1.x版本之前通过指定max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）
• 1.x版本之后未开启幂等性，需要设置max.in.flight.requests.per.connection=1；开启幂等性后，max.in.flight.requests.per.connection<=5(启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，无论如何，都可以保证近5个request的数据都是有序的，通过幂等性里面的SeqNumber判断有序)

10、幂等性原理

• 幂等性： Producer不论向Broker发送多少次重复数据，Broker都只会持久化1条，保证了不重复（仅保证单分区会话内不重复）
• 精确一次： 幂等性 +至少一次（ACK=-1 + 分区副本数>=2 + ISR最小副本数量>= 2）
• 重复数据判断标准： 具有<PID，Partition，SeqNumber>相同主键消息提交时,Broker只会持久化一条，PID是kafka进程号，Partition标识分区号，Sequence Number单调自增
• 开启条件： 通过参数enable.idempotence 开启，默认true

11、为什么kafka放弃zookeeper

• 从系统运维角度来说kafka本身是一个分布式系统，其自身运维已经十分复杂；同时依赖zookeeper，增加运维成本
• 对于kafka自身的发展而言，过度依赖外部系统不利于自身功能的完善与维护
• 在性能方面，offset、isr等信息保存在zookeeper中，zookeeper不适合高频的读写与更新操作，会影响其性能，新版本的kafka已经将提交和保存用消息的方式进行存储
• zookeeper已经成为kafka的瓶颈

12、kafka文件存储机制

• 1、Topic是逻辑上的概念，Partition是物理上的概念，每个Partition对应一个log文件（即生产数据）。Partition产生的数据会被追加到log文件末尾（顺序写入）。
• 2、为了防止log过大导致数据定位效率低下，kafka采用分片和索引机制，将每个Partition分割成多个Segement，每个Segement包括“.index”文件，“.log”文件和“.timeindex”等文件，统一位于topic+分区编号目录下
  • 注：index和log文件以当前segement的第一条消息的offset命名
• 3、Log&Index：
  • index为稀疏索引 ，大约每往log写入4kb数据，往index写入一条索引，参数log.index.interval.bytes默认4kb
  • index文件中保存offset为相对offset，这样确保offset的值占用空间不会过大，因此将offset的值控制在固定大小（真实offset=文件名数据+文件内offset）
• 4、文件清理策略
  • 清理周期： 默认日志保存时间为7天 ，还可以设置清理间隔为分钟、毫秒，优先级：小时<分钟<毫秒；检查周期默认5mins
  • delete策略：
    • 基于时间：默认打开，以segement中所有记录的最大时间戳作为文件的时间戳
    • 基于大小：默认关闭，超过设置日志总大小，删除最早的segement
  • compat策略：
    • 日志压缩：对相同key的不同value，保留最后一个版本
    • 压缩后offset可能不连续
    • 此策略适合场景：消息key是用户ID，value是用户资料，通过这种策略，保存的用户数据一直是最新数据

13、产生数据积压，如何快速处理

• 1、增加分区Partitions数，一般不能大于kafka的broker数；如果大于，会出现无法消费数据的情况
• 2、提高消费者的速度，增大每次拉取的缓冲区数据的最大值（batch.size），或者加大每次拉取的数据条数
• 3、调整生产者的参数，适当增加缓冲区一批数据的大小（batch.size）；增大延迟时间（linger.ms），默认0ms，标识没有延迟，生产环境建议5-100ms； 压缩生产者发送的数据（compression.type），默认none(不压缩)，支持压缩类型gzip，snappy，lz4，zstd（压缩比高的压缩时间长，压缩比低的效果不明显）

14、硬件估算

• 服务器台数=2*（生产者分支生产率 * 副本/100） +1
• cpu数量：
  • num.io.thread：写磁盘线程，占总核数50%
  • num.replica.fetchers：副本拉取线程数，占总核数50%的1/3
  • num.network.threads：数据传输线程，占总核数50%的2/3

15、kafka为什么比较快

高效读写数据

• 分布式集群： kafka本身是分布式集群，可采用分区技术，并行度高
• 稀疏索引： 读数据采用稀疏索引，能够快速定位到要消费的数据
• 顺序IO： kafka写数据到分区采用追加的方式（顺序写入），所以这个速度非常快；物理上采用磁盘和固态硬盘区别不大；官网数据：顺序读写600M/s,随机只有100K/s
• 零拷贝和页缓存： kafka在写入消息的时候通过mmap内存映射的方式，上层有写操作的时候，操作系统只将数据写入pageCache，发生读操作时，先从pageCache查找，找不到再去磁盘查找，实际上pageCache尽可能多的将空闲内存作为磁盘缓存使用； kafka的数据加工由生产者和消费者处理，broker应用不关心存储数据，所以数据不走应用层，通过sendfile实现零拷贝，将数据直接发送到consumer
• 批处理和压缩： kafka在发送和消费消息的时候，是一批一批处理数据；且可对消息进行压缩，缩小消息体积，可以一次性传输更多消息