kafka记录及面试题

忘是亡心i 2022-11-25 14:29 179阅读 0赞

1、kafka的介绍：

分布式发布-订阅消息队列，生产者往队列里写消息，消费者从队列里取消息进行业务逻辑。一般在架构设计中起到**解耦（**允许我们独立的扩展或修改队列两边的处理过程**）、异步处理（**允许用户把消息放入队列但不立即处理它**）、缓冲削峰（其实就是**解决生产消息和消费消息的处理速度不一致的情况**）**的作用。

2、基本概念：

消费者组CG：消费者组内每个消费者负责消费不同partition的数据，提高消费能力。一个分区只能由同一组内一个消费者消费，消费者组之间互不影响，消费者组是逻辑上的一个订阅者。

Broker：一台 Kafka 机器就是一个 Broker。

Partition**：**是为了实现扩展性，提高并发能力，一个 Topic 可以分为多个 Partition分布在不同的broker上，每个 Partition 是一个 有序的队列。

Topic ：是逻辑上的概念，而 Partition 是物理上的概念，每个 Partition 对应着一些log 文件，Producer 生产的数据会不断append追加到 log 文件末端，且每条数据都有自己的 Offset，且消费者组中的每个消费者，都会实时记录自己消费到了哪个 Offset，以便出错恢复时，从上次的位置继续消费。生产者的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了partition分片和索引机制，将每个 Partition 分为多个 Segment，每个 Segment 对应一个“.index” 索引文件和 “.log” 数据文件，“.index” 文件存储.log文件中 Message 的物理偏移量，这些文件位于同一文件下命名规则为topic 名-分区号，index 和 log 文件以当前 Segment 的第一条消息的 Offset 命名。

Offset：Kafka中的每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中，partition中的每个消息都有一个连续的序号，也就是offset用于为partition唯一标识一条消息。Offset从语义上来看拥有两种：Current Offset和Committed Offset。Current Offset是保存在Consumer客户端中，是针对Consumer的poll过程的，它可以保证每次poll都返回不重复的消息；Committed Offset是用于Consumer Rebalance过程的，它能够保证新的Consumer能够从正确的位置开始消费一个partition，从而避免重复消费。在Kafka 0.9前，Committed Offset信息保存在zookeeper的（zookeeper其实并不适合进行大批量的读写操作，尤其是写操作），因此在0.9之后，所有的offset信息都保存在了Broker上的一个名为\_\_consumer\_offsets的topic中。

`Group Coordinator：`消费过程中，每个 `Broker都会`在启动时启动一个`Group Coordinator` 服务，CG确定自己属于哪一个`Coordinator` ，首先确定该CG的offset信息写入到\_\_consumers\_offsets topic中的分区，该分区的leader所在的broker就是CG所属的coordinator。

Group Coordinator都会存储消费者组以及组内消费者的配置信息，主要包括：订阅的topics列表；Consumer Group配置信息，包括session timeout等；组中每个Consumer的元数据。包括主机名，consumer id；每个Group正在消费的topic partition的当前offsets；Partition的ownership元数据，包括consumer消费的partitions映射关系。

3、分区策略

（1）msg->partition

将 Producer 发送的msg封装成一个 ProducerRecord 对象，对该对象指定一些参数后，然后进行序列化，然后按照topic和partition放进对应的发送队列中，这里有异步和同步两种发送方式。具体参数包括：

**topic：**string 类型，NotNull；**partition：**int 类型，可选；**timestamp：**long 类型，可选；**key：**string 类型，可选；**value：**string 类型，可选；**headers：**array 类型，Nullable。

①指明 Partition 的情况下，直接将给定的值作为 Partition 的值。

②没有指明 Partition 但有 Key 的情况下，将 Key 的 Hash 值与分区数取余得到 Partition 值。

③既没有 Partition 也没有 Key 的情况下，Round-Robin 轮询算法，第一次调用时随机生成一个整数（后面每次调用都在这个整数上自增），将这个值与可用的分区数取余，得到 Partition 值。

（2）patition->broker

首先将所有Broker和待分配的Partition排序；其次将第i个Partition分配到第（i mod n）个Broker上 （这个就是leader）；最后将第i个Partition的第j个Replica分配到第（(i + j) mod n）个Broker上。

kafka使用zookeeper在broker中选出一个**controller**，用于**partition在broker的分配和partition leader选举**。

4、数据可靠性保证

（1）冗余策略

分布式架构都会设置冗余副本数，kafka中partition的冗余机制默认是一个leader和两个follower，所有对该partition的操作，实际操作的都是leader，然后再同步到其他的follower，follower是从leader批量拉取数据来同步，并且会尽量把多个副本，分配到不同的broker上。

（2）ack应答机制

提供了三种可靠性级别，用户根据可靠性和延迟的要求进行权衡，Ack 参数配置：

**0：**Producer 不等待 Broker 的 ACK，这提供了最低延迟，Broker 一收到数据还没有写入磁盘就已经返回，当 Broker 故障时有可能丢失数据。

**1：**Producer 等待 Broker 的 ACK，Partition 的 Leader 落盘成功后返回 ACK，如果在 Follower 同步成功之前 Leader 故障，那么将会丢失数据。

**-1（all）：**Producer 等待 Broker 的 ACK，Partition 的 Leader 和 Follower 全部落盘成功后才返回 ACK。但是在 Broker 发送 ACK 时，Leader 发生故障，则会造成数据重复。如果设置ack=-1，设想有一个 Follower 因为某种原因出现故障，那 Leader 就要一直等到它完成同步，Leader维护了一个动态的ISR，如果 Follower 长时间未向 Leader 同步数据，则该 Follower 将被踢出 ISR 集合，Leader 发生故障后，controller就会从 ISR 中选举出新的 Leader。

（3）故障处理

LEO：每个partition最大的 Offset。HW：消费者能见到的最大的 Offset，即ISR 队列中最小的 LEO。

**Follower 故障：**Follower 发生故障后会被临时踢出 ISR 集合，待该 Follower 恢复后，Follower 会 读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 Leader 进行同步数据操作，等该 Follower 的 LEO 大于等于该 Partition 的 HW，即 Follower 追上 Leader 后，就可以重新加入 ISR 了。

**Leader 故障：**Leader 发生故障后，controller会从 ISR 中选出一个新的 Leader，之后，为保证多个副本之间的数据一致性，其余的 Follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 Leader 同步数据，这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

（4）Exactly Once 语义

将服务器的 ACK 级别设置为 -1，可以保证 Producer 到 Server 之间不会丢失数据，即 At Least Once 语义，但是不能保证数据不重复。

将服务器 ACK 级别设置为 0，可以保证生产者每条消息只会被发送一次可以保证数据不重复，即 At Most Once 语义，但是不能保证数据不丢失。

但是，对于一些非常重要的信息比如交易数据，下游数据消费者要求数据既不重复也不丢失，即 Exactly Once 语义。

0.11 版本的 Kafka，引入了幂等性：Producer 不论向 Server 发送多少重复数据，Server 端都只会持久化一条。

即：**At Least Once + 幂等性 = Exactly Once**

开启幂等性的 Producer 在初始化时会被分配一个 PID，发往同一 Partition 的消息会附带 Sequence Number，而 Borker 端会对 **<PID,Partition,SeqNumber>** 做缓存，当具有相同值的消息提交时，Broker 只会持久化一条，但是 PID 重启后就会变化，且不同的 Partition 也具有不同主键，所以幂等性无法保证跨分区跨会话的 Exactly Once。

5、消费方式

Consumer 采用 Pull（拉取）模式从 Broker 中读取数据，可以根据 Consumer 的消费能力以适当的速率消费消息。Pull 模式不足之处是，如果 Kafka 没有数据，消费者可能会陷入循环一直返回空数据。针对这一点， Kafka 的消费者在消费数据时会传入一个时长参数 timeout，如果当前没有数据可供消费，Consumer 会等待timeout时间之后再返回。

Consumer 采用 Push（推送）模式，Broker 给 Consumer 推送消息的速率是由 Broker 决定的，很难适应消费速率不同的消费者。目标是尽可能以最快速度传递消息，但是这样很容易造成 Consumer 来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。

6、reblance消费者组内分区分配策略

发生场景：消费者组成员个数发生变化；订阅的 `Topic` 个数发生变化；订阅 `Topic` 的分区数发生变化。

（1） RoundRobin，轮询方式将该CG订阅的所有topic的所有分区作为一个整体进行 Hash 排序，消费者组内分配分区个数最大差别为 1，是按照组来分的，可以解决多个消费者消费数据不均衡的问题。但是当消费者组内订阅不同主题时，可能造成消费混乱。

（2） Range，默认为Range，不会产生消费混乱问题，其实就是单个topic的轮询分配，但是同时订阅了主题 A 和 B，可能造成消息分配不对等问题，当消费者组内订阅的主题越多，分区分配可能越不均衡。

`Rebalance` 发生时，`Group` 下所有 `Consumer` 实例都会协调在一起共同参与，`Kafka` 能够保证尽量达到最公平的分配。但是 `Rebalance` 过程对 `Consumer Group` 会造成比较严重的影响。在 `Rebalance` 的过程中 `Consumer Group` 下的所有消费者实例都会停止工作，等待 `Rebalance` 过程完成。

**常见面试题：**

1、Kafka中的ISR、AR又代表什么？ISR的伸缩又指什么

ISR:In-Sync Replicas 副本同步队列  
AR:Assigned Replicas 所有副本  
ISR是由leader维护，每个Partition都会有一个ISR，follower从leader同步数据有一些延迟，包括延迟时间和延迟条数两个维度，任意一个超过阈值都会把follower剔除出ISR, 存入OSR（Outof-Sync Replicas）列表，新加入的follower也会先存放在OSR中。AR=ISR+OSR。

2、什么是broker

broker 是消息的代理，Producers往Brokers里面的指定Topic中写消息，Consumers从Brokers里面拉取指定Topic的消息，然后进行业务处理，broker在中间起到一个**代理保存消息的中转站**。

**3、kafka follower如何与leader同步数据**

Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。完全同步复制要求All Alive Follower都复制完，这条消息才会被认为commit，这种复制方式极大的影响了吞吐率。而异步复制方式下，Follower异步的从Leader复制数据，数据只要被Leader写入log就被认为已经commit，这种情况下，如果leader挂掉，会丢失数据，kafka使用**ISR的方式**很好的均衡了确保数据不丢失以及吞吐率。Follower可以**批量**的从Leader复制数据，而且Leader充分利用**磁盘顺序读**以及**send file(zero copy)机制**，这样极大的提高复制性能，内部批量写磁盘，大幅减少了Follower与Leader的消息量差（**顺序写、零拷贝、批量量处理也是kafka快的原因**）。

**4、什么情况下一个 partition或broker 会从 isr中踢出去**

leader会维护一个副本同步列表ISR(in-sync Replica)，每个Partition都会有一个ISR，而且是由leader动态维护 ，如果一个follower比一个leader落后太多，或者超过一定时间未发起数据复制请求，则leader将其从ISR中移除 。

**5、kafka producer 打数据，ack  为 0， 1， -1 的时候代表啥， 设置 -1 的时候，什么情况下，leader 会认为一条消息 commit了**

1（默认）  数据发送到Kafka后，经过leader成功接收消息的的确认，就算是发送成功了。在这种情况下，如果leader宕机了，则会丢失数据。

0 生产者将数据发送出去就不管了，不去等待任何返回。这种情况下数据传输效率最高，但是数据可靠性确是最低的。

\-1 producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。当ISR中所有Replica都向Leader发送ACK时，leader才commit，这时候producer才能认为一个请求中的消息都commit了。

**6、如果leader crash时，ISR为空怎么办**

kafka在Broker端提供了一个配置参数：unclean.leader.election,这个参数有两个值：  
true（默认）：允许不同步副本成为leader，由于不同步副本的消息较为滞后，此时成为leader，可能会出现消息不一致的情况。  
false：不允许不同步副本成为leader，此时如果发生ISR列表为空，会一直等待旧leader恢复，降低了可用性。

**7、kafka的message格式是什么样的**

一个Kafka的Message由一个固定长度的header和一个变长的消息体body组成

header部分由一个字节的magic(文件格式)和四个字节的CRC32(用于判断body消息体是否正常)构成。

当magic的值为1的时候，会在magic和crc32之间多一个字节的数据：attributes(保存一些相关属性，

比如是否压缩、压缩格式等等);如果magic的值为0，那么不存在attributes属性

body是由N个字节构成的一个消息体，包含了具体的key/value消息

**8、kafka中consumer group 是什么概念**

同一个topic的数据，会广播给不同的消费者组；同一个group中只有一个worker能拿到这个数据进行消费。group内的worker可以使用多线程或多进程来实现，也可以将进程分散在多台机器上，一个组内worker的数量通常不超过partition的数量，且二者最好保持整数倍关系。

**9、幂等性**

即f(x)=f(f(x)) 能够成立的数学性质。用在编程领域，则意为`对同一个系统，使用同样的条件，一次请求和重复的多次请求对系统资源的影响是一致的`

**10、Kafka中的消息是否会丢失和重复消费？**

要确定Kafka的消息是否丢失或重复，从两个方面分析入手：消息发送和消息消费。

**1、消息发送**

Kafka消息发送有两种方式：同步（sync）和异步（async），默认是异步方式，可通过producer.type属性进行配置。Kafka通过配置request.required.acks属性来确认消息的生产：

1.  *0---表示不进行消息接收是否成功的确认；*
2.  *1---表示当Leader接收成功时确认；*
3.  *-1---表示Leader和Follower都接收成功时确认；*

综上所述，有6种消息生产的情况，下面分情况来分析消息丢失的场景：

（1）acks=0，不和Kafka集群进行消息接收确认，则当网络异常、缓冲区满了等情况时，**消息可能丢失**；

（2）acks=1，同步模式下，只有Leader确认接收成功后但挂掉了，副本没有同步，**消息可能丢失**；

**2、消息消费**

Kafka消息消费有两个consumer接口，Low-level API和High-level API：

1.  Low-level API：消费者自己维护offset等值，可以实现对Kafka的完全控制；
2.  High-level API：封装了对parition和offset的管理，使用简单；

如果使用高级接口High-level API，可能存在一个问题就是当消息消费者从集群中把消息取出来、并提交了新的消息offset值后，还没来得及消费就挂掉了，那么下次再消费时之前没消费成功的消息就“*诡异*”的消失了；

**解决办法**：

针对消息丢失：同步模式下，确认机制设置为-1，即让消息写入Leader和Follower之后再确认消息发送成功；异步模式下，为防止缓冲区满，可以在配置文件设置不限制阻塞超时时间，当缓冲区满时让生产者一直处于阻塞状态；

针对消息重复：将消息的唯一标识保存到外部介质中，每次消费时判断是否处理过即可。

**11、如何保证消息不被重复消费？或者说，如何保证消息消费时的幂等性？**

Kafka 实际上有个 offset 的概念，就是每个消息写进去，都有一个 offset，然后 consumer 消费了数据之后，每隔一段时间会把自己消费过的消息的 offset 提交一下，表示我已经消费过了,但是如果进程直接挂掉了，重启后这会导致 consumer 有些消息处理了但是没来得及提交 offset，少数消息会再次消费一次。

其实重复消费不可怕，可怕的是你没考虑到重复消费之后，**怎么保证幂等性**。

举个例子吧。消费一条往数据库里插入一条，要是你一个消息重复两次，就插入了两条数据就出错了，但是你要是消费到第二次的时候，自己判断一下已经消费过了，直接扔了，不就保留了一条数据，一条数据重复出现两次，数据库里就只有一条数据，这就保证了系统的幂等性。

如何保证幂等性主要是要结合具体的业务：

如果是数据要写库，你先根据主键查一下，如果这数据都有了，你就别插入了，update 一下好吧。

如果是写 Redis，那没问题了，反正每次都是 set，天然幂等性。

如果更复杂一点，你需要让生产者发送每条数据的时候，里面加一个全局唯一的 id存在redis中，消费时判断一下id是否消费过。

12、**Kafka中是怎么体现消息顺序性的？**

kafka每个partition中的消息在写入时都是有序的，消费时每个partition只能被每个group中的一个消费者消费，保证了消费时也是有序的。topic不保证有序。如果要保证topic整个有序，那么将partition调整为1.