Kafka

应用场景

消息队列常见应用场景有哪些？

答：

示例1（概括性问题）
- 消息队列定位就是一些特定场景下解决消息传输问题，比如：异步场景、服务间解耦场景、消息分发场景，以及压力过大时可以通过消息队列削峰
示例2（列举回答）
- 应用场景很多，比如如下几点：
  1. 系统解耦：在重要操作完成后，发送消息到 Kafka 中，由别的服务系统来消费消息完成其他操作
  2. 流量削峰：一般用于秒杀或抢购活动中，来缓冲系统短时间内高流量带来的压力
  3. 异步处理：通过异步处理机制，可以把一个消息放入队列中，但不立即处理它，在需要的时候再进行处理
  4. 消息分发：比如一条消息需要传递给多个服务，这时候就可以用 Kafka 进行分发

什么情况下需要解耦？

答：

比如：发送短信场景，模块 A 发送消息给 B，模块 B 发送短信给客户，A 不需要得到回应，对于 A 而言只需要触达 B 就行了，这时候就可以引入消息队列，A 将消息投递到了消息队列，B 自己去处理，A 不用再关心。这样可以收获更强的稳定性以及接口性能

补充：

解耦是一个比较抽象的词，用一个例子来讲述会比较容易

什么情况下需要削峰？

答：

比如：一个模块 B 一瞬间收到 100 个请求，如果 B 的承压能力非常差，或者 B 有什么资源限制，那么这 100 个请求下来，B 可能就挂了或者报错了，这时候就可以引入消息队列，前置服务先将消息打到消息队列，B 再根据自己的消费能力逐步消化

补充：

压力过大扛不住、接入消息队列后可按需消费

什么情况下需要分发消息？

答：

比如：信息更新场景，某个用户信息更新了，而 B、C、D 三个模块都需要缓存这个信息，那么用户信息更新之后就可以发一条信息到消息队列，B、C、D 只要订阅了相关主题，就都可以获得这条信息并做对应的缓存更新

补充：

一个消息需要传递给多个接收者的场景，一般而言还是举个例子去说明

在项目开发中，你会怎么选择消息队列？

答：

我会对比常见的几个消息队列，从 功能性、性能、健壮性 上去做对比，比如：如果我们的性能要求非常高，我就会偏向于 Kafka 和 RocketMQ 这类性能又高，又自带扩展性的消息队列。当然还有一个考虑点就是团队技术栈，如果基本要求都满足，会更倾向于选择团队使用成熟的组件

补充：

要点：要说到从需求（功能性）上考虑，以及性能和稳定性。如果有实战经验的同学也可以说一下基本熟悉成熟做选择，没实战经验的同学可以不一定能完全 GET 到熟悉是多重要的事情，如果不能很自信的说出来，那不如不说

Kafka 和 RocketMQ 有什么区别？

答：

例一：
- 我之前主要都是在研究/使用 Kafka，RocketMQ 只有一些粗浅的认知，比如：RocketMQ 可以支持更多的主题，适合于需要非常多主题的场景，以及 RocketMQ 的功能应该更完善一些，比如：天然支持死信队列
例二（熟悉RocketMQ）：
- RocketMQ 和 Kafka 相比，在架构上做了减法，在功能上做了加法
- 跟 Kafka 的架构相比，RocketMQ 简化了协调节点和分区以及备份模型。同时增强了消息过滤、消息回溯和事务能力，加入了延迟队列，死信队列等新特性

详细链接：RocketMQ 是什么？它的架构是怎么样的？和 Kafka 又有什么区别？

Kafka 的优势是什么？

答：

Kafka 优点非常多，我认为最核心的是高吞吐、高可靠、天然支持分片水平扩展，另外还有一个非常重要的考量就是 Kafka 多年来被广泛使用，并且社区非常活跃，这就意味着它是经过市场验证的好产品

补充：

Kafka 最核心的三点优势：高性能、天然水平扩展、被广泛使用

服务端

Kafka 的大概框架是怎么样的？

答：

可以先把 Kafka 宏观看作三层：Producer（生产者）、Server（中转者）、Consumer（消费者），生产者发送消息，服务端负责存储消息，消费者负责拉取消息。其中：服务端其实就是由多个 Broker 节点组成，而平常说的主题就是在 Broker 节点上，当然，Topic 是个逻辑概念，实际物理存储形式是主题分片，也就是Partition

Kafka大致框架

补充：

Broker：可以理解为机器或者节点吧，也可以理解为就是运行 Kafka 程序的服务器
Topic：主题是 Kafka 中的一个核心概念，它是对消息进行分类的一种方式。生产者将消息发送到特定的主题中，而消费者则通过订阅主题来接收相关的消息。但是要注意的是，主题是一个 逻辑概念，实际上，一个主题可以被分为多个分区（Partition），以实现消息的并行处理和负载均衡，数据是存储在Partition 这个级别的
Partition：分区是 Kafka 中的一个重要概念，它是 主题的物理存储单位。每个分区都是一个有序的、不可变的消息序列，可以被独立地读写。分区在物理上对应一个文件夹及文件夹下面的文件，分区的命名规则为主题名称后接 - 连接符，之后再接分区编号，比如：TopicA-1 就表示主题A 的 1 号分区，每个分区又可以有一至多个副本（Replica）以提高可用性

整体框架

生产消费：生产者 Producer，消费者 Consumer
存储：节点 Broker，逻辑主题 Topic，物理分片 Partition

如何获取 topic 主题的列表？

答：

Kafka 是提供了获取主题列表的接口的，可以使用 kafka-topics.sh 这个工具获取，如果要在我们业务服务来获取的话，那主流语言都是支持获取的，比如：Java和Golang，都可以用 KafkaAdminClient 直接调用接口来获取

补充：

Kafka 是提供了获取主题列表的接口的，如果是使用自带工具，可以用这个命令获取：./kafka-topics.sh --ist --bootstrap-server localhost:9092

Java 和 Golang 都是用 AdminClient 获取，Java 例子：

public class KafkaTopicLister {
	public static void main(string[] args){
		// Kafka broker 地址
    	String brokerAddress ="localhost:9092";
		// 配置 KafkaAdminclient
		Properties properties = new Properties();
		properties.put(AdminclientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerAddress);
		try(Adminclient adminclient = Adminclient.create(properties)){
        	// 获取主题列表
			ListTopicsoptions options = new ListTopicsoptions();
			options.listInternal(false); //是否列出内部主题
			adminclient.listTopics(options).names().get().forEach(system.out::println);
    	}catch(InterruptedException | ExecutionException e) {
        	e.printstackTrace();
    	}
    }
}

有了Topic，为何还要 Partition？

Kafka 为什么要把消息分区？

答：

例一（简洁）：
- 一般而言一个业务可以用一个主题，但是就算分了 Topic ，单个业务的信息还是可能会非常多，所以需要能进一步进行分治，也就是一个主题由多个 Partition 组成，这样相同的主题也具备了更高的并发度
例二（细致一点）：
- 如果不进行分区，那么我们发消息写数据都只能保存到一个节点上，这样的话可能会导致单点服务器负载过高的问题，通过分区可以把数据均匀地分布到不同的节点。因此，分区带来了负载均衡和横向扩展的能力
- 发送消息时可以根据分区分配的原则落在不同的 Kafka 服务器节点上，提升了并发写消息的性能，消费消息的时候又和消费者绑定了关系，可以从不同节点的不同分区消费消息，提高了读消息的能力
- 另外一个就是分区又引入了副本，冗余的副本保证了 Kafka 的高可用和高持久性

补充：

Kafka 消息会分发到不同 Topic，这样既解决了消息混乱的问题，又顺带将流量进行了分摊，减少单个业务写入压力
但是，不同业务对应的数据量级是不一样的，就算按业务分了 Topic，但一个 Topic 可能信息还是会非常多，我们需要进一步分而治之，所以 Topic 下面还划分了 Partition 来实际存储，每个 Topic 都可以有多个 Partition

Partition 是逻辑概念还是物理概念？

答：

Partition 是物理概念，因为数据实实在在是写入到 Partition 文件里去的。而因为设计了 Partition 的存在，Topic 就只是一个逻辑概念了

补充：

逻辑概念就是说是一个抽象概念，没有实际体现在物理存储上，物理概念则是相反的，Topic 只是逻辑概念，Partition 则是一个实实在在的物理存储概念

介绍一下分区的分配策略？

答：

Range Assignor：基于范围的分配策略，将分区按照范围分配给消费者
RoundRobin Assignor：基于轮询的分配策略，分区均匀地分配给消费者
Sticky Assignor：优先保持当前的分配状态，并尽量减少在再平衡过程中的分区移动
CooperativeStickyAssignor：和 Sticky Assignor 的策略是基本一样的，区别在于该协议将原来的一次大的全部分区重平衡，改成多次小规模分区重平衡。简单理解就是 渐进式重平衡

补充：

我们可以关注消费端 partition.assignment.strategy 这个配置，它有如下几种选择：

Range Assignor：基于范围的分配策略，将分区按照范围分配给消费者
RoundRobin Assignor：基于轮询的分配策略，分区均匀地分配给消费者
Sticky Assignor：优先保持当前的分配状态，并尽量减少在再平衡过程中的分区移动
CooperativeStickyAssignor：和 Sticky Assignor 的策略是一样，区别在于未受变动的消费者可以继续消费主题

Kafka 创建 Topic 时如何将分区放置到不同的 Broker 中？

答：

具体而言某个 Topic 的 Partition 分配的规则如下，我们以 2 个 Partition 的 TopicB 举例：

先随机选取一个 Broker，比如：Broker11
将主题对应的第一个分片，放入 Broker11，即：TopicB 的 0 号分片放到了 Broker11
依次往后放 TopicB 后续的分片，比如：TopicB 的 1 号分片放入了 Broker12，TopicB 在我们的例子中只有 2 个 Partition，假设有 3 个，那么下一个就放到 Broker10

总结一下规则：就是 kafka 是先随机挑选一个 broker 放置分区 0，然后再按顺序放置其他分区

补充：

Partition 是存放在 Broker 节点上的，如果单个 Broker，很好理解，Partition 都在 Broker 上。如果是多个 Broker，则 Partition 会分布到不同的 Broker上，简单来说，一个 Partition 只对应一个 Broker，一个 Broker 可以存放多个 Partition

Broker节点分区

消息存入哪个 Partition 的规则？

Kafka 中分区分配的原则？

答：

一个主题的数据分散成了多个分片，我们就需要有一种方式来决定消息是写入哪个分片，规则如下：

如果指定了 Partition，那么就是发送到特定的 Partition，但是一般情况下，业务其实不需要感知Partition，除非有特殊理由，否则不建议直接指定要发送到哪个 Partition
如果没有指定 Partition，但是指定了一个 Key，那么就是根据 Key 的 Hash 对 Partition 数目取模来决定是哪个 Partition，也就是说只要发送时指定了相同的 Key，那么相关消息一定会发送到相同的 Partition
如果没有指定 Partition，也没有指定 Key，那么就采取轮询调度算法，也就是每一次把来自用户的请求轮流分配给 Partition

Kafka 服务端可接收的消息最大默认多少字节，如何修改？

答：

Kafka 可以接收的最大消息默认为 1MB，如果想调整它的大小，可在Broker中修改配置数：message.max.bytes 的值

Kafka 的 Topic 中 Partition 数据是怎么存储到磁盘的？

答：

Topic 中的多个 Partition 以 文件夹 的形式保存到 Broker，每个分区序号从 0 递增，且消息有序
Partition 文件下有多个 Segment(xxx.index, xxx.log)，Segment 文件的大小是可以进行配置的。默认为 1GB。如果大小大于 1GB 时，会滚动一个新的 Segment 并且以上一个 Segment 最后一条消息的偏移量命名

Kafka如何清理数据？

Kafka数据越积越多怎么办？

答：

可以用 基于时间的保留策略，这种策略允许用户指定消息的保留时间（如：7天）。超过指定时间的消息将被自动删除
也可以用 基于大小的保留策略，Kafka 允许用户指定日志的最大尺寸。一旦日志的大小超过了配置的值，Kafka 将开始删除最早的消息

生产者

介绍一下生产消息的流程？

答：

生产消息是发生在业务服务器那一侧，看似简单，其实也分了好几步来做
1. 第一步是 构建消息，即：将要发送的内容，打包成一个 Kafka 的消息结构
2. 第二步是 序列化消息为二进制内容，以在网络中传输
3. 第三步是 进行分区选择，即：计算要发到哪个 Partition，发送消息到该 Partition 对应的 Broker

补充：

生产者是在业务服务器那一侧
讲述生产消息的流程

讲一讲 kafka 的 ack 的三种机制？

答：

第一种模式，ack = 0，生产者在发送消息后不会等待来自服务器的确认，所以生产者实际是不知道消息是否成功,也就无从去重试，生产可靠性是最低的
第二种模式，ack = 1，生产者会在消息发送后等待主节点的确认，但不会等待所有副本的确认
第三种模式，ack = all，只有在所有副本都成功写入消息后，生产者才会收到确认。这确保了消息的可靠性，但延迟显然是最高的

补充：

acks	行为	可靠性	性能
`0`	生产者在发送消息后不会等待来自服务器的确认，所以生产者实际上不知道消息是否成功，也就无法重试，生产可靠性是最低的	不可靠	最高
`1`	生产者会在消息发送后等待主节点的确认，但不会等待所有副本的确认	相对可靠	还是比较高
`all` 或 `1`	只有在所有副本都成功写入消息后，生产者才会收到确认。这确保了消息的可靠性，但延迟显然是最高的	可靠	会低一些

图示解析

生产过程中何时会发生 QueueFullExpection 以及如何处理？

答：

生产者发送消息的速度过快，导致生产者这一侧缓冲区满了，有几条路去解决：
1. 等待重试：当发生 QueueFullException 异常时，可以等待一段时间后再次尝试发送消息。在等待的过程中，可以通过调整生产者的发送速度或者增加 Kafka 的消息队列大小等方式来避免再次发生QueueFullException 异常
2. 增加 Kafka 的缓冲区大小：可以通过修改 Kafka 的配置文件来增加缓冲区的大小，以适应生产者发送消息的速度
3. 限流控制：可以通过限制生产者发送消息的速度来避免 QueueFullException 异常的发生

补充：

生产者发送消息的速度过快，导致生产者这一侧缓冲区满了，就会抛出 QueueFullExpection 这个错误

Kafka 生产者何时发出消息？

答：

累计的数据大小达到 Batch 大小，默认：16KB
缓冲区中累计的空闲等待时间间隔，默认 0ms，也就是默认收到数据就发送

通过增加 Batch 的大小，可以减少网络请求和磁盘 I/O 的频次，具体参数配置需要在效率和时效性做一个权衡

生产者发送消息的模式有几种？（Java 版本）

答：

有三种模式：同步发送、发送即忘、异步发送
- 同步发送 性能最差，但是可靠性最强
- 发送即忘 性能最高，这种模式不需要等待 Kafka 服务器的响应，所以可靠性低，也不知道是不是发送成功了
- 异步发送 不阻塞调用线程，同时允许调用者处理发送结果或异常。适用于对消息传输可靠性有要求，同时希望保持高性能的场景
其中异步发送模式因为 兼顾（折中）了可靠性和性能，所以是最被广泛使用的

补充：

Java 的 SDK 支持三种发送模式：

同步发送，这种方式会阻塞调用线程，直到消息发送成功或抛出异常。适用于需要确保消息发送成功后才进行后续处理的场景。这种速度会慢一些，但是可以很准确知道发送是否成功
1
2
3
4
5
try {
kafkaTemplate.send("topicName", "message").get();
} catch (InterruptedException | ExecutionException e) {
e.printstackTrace();
}
发送即忘（Fire-and-forget）发送了就不管了，这种方式是最简单的发送方式。消息发送后不等待任何响应，也不处理可能的异常
1
kafkaTemplate.send("topicName", "message");

异步发送，发送了就做其它事情去了，这种方式不会阻塞调用线程，而且可以 注册回调函数来处理发送结果或异常

ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("topicName", "message");
future.addcallback(new Listenablefuturecallback<sendResult<String, String>>() {
    @Override
    public void onSuccess(SendResult<String, String> result) {
        System.out.println("Sent message = [" message + "] with offset = [" result.getRecordMetadata().offset()+ "]");
    }
    
    @Override
    public void onFailure(Throwable ex) {
        System.out.println("Unable to send message = [" + message + "] due to :" ex.getMessage());
    }
});

消费者

Kafka 消费者是推模式还是拉模式？

Kafka 消息的消费模式？

答：

Kafka 采用 拉模式 拉取消息，采用拉模式可以使每个消费者以自身的消费能力去消费。拉模式有个缺点是，如果 Broker 没有可供消费的消息，将导致 Consumer 不断在循环中轮询，直到新消息到达。为了避免这点，Kafka 消费者可以使用在消费数据时传入 timeout 参数，在这个时间范围内进行阻塞等待，直到有数据或超时后再返回

补充：

Kafka 的消费者使用的拉模式来获取信息，也就是说每次消费者是发消息到 Kafka 的 Broker 来获取信息，而不是由 Kafka 的 Broker 主动推送
选择拉模式的主要原因还是为了让 消费者可以按自身情况来控制消费速度，根据系统资源利用情况（如：CPU、内存等）、业务需要等因素合理拉取消息，避免因消息处理速度不合理带来的资源浪费或过载

消费者故障，出现活锁问题如何解决？

答：

可以使用 最大拉取间隔 这个参数来解决活锁问题，即：max.poll.interval.ms，如果消费者轮询间隔大于了这个值，消费者就会离开分区，这样其它消费者就可以接管对应分区

补充：

活锁：消费者持续的维持心跳，但因为异常进行消息处理，或者消息处理非常长时间卡住了，这种情况下，这个消费者在就会一直持有分区，该分区的消息就无法得到处理，要解决这个问题，需要有活锁检测机制

有消费者为什么还要消费者组？

答：

消费组其实就是把多个消费者组织在一起进行消费。我觉得解决了几个问题吧：

对主题分片的分配问题，让每个分片都能有消费者处理，又不至于所有消费者处理同一个分片
面对主题分片的变化，消费组可以自动调整，也就是再平衡
对于业务开发者而言，有了消费组，就只用关心主题维度，而不用关心分片维度，很大程度降低了理解和应用难度

补充：

Consumer Group：

消费组其实就是把一些消费者组织起来，一同工作，它们像一个团队一样，协同作战

实现指令：

1	sh kafka-console-consumer.sh --bootstrap-server localhost:9092 --group xxx --topic testxxx

一个消费组由一个唯一的 groupid 来标识，这个 groupid 是在消费者这一侧指定的，拥有相同 groupid 的消费者就属于同一个消费组
具体而言消费组做这么一件事：
- 一个消费组中的消费者消费同一个 Topic，每个消费者可以承接这个 Topic 一部分 Partition 的消息，这样消费能力就实现了水平扩展
- 注意：
  1. 同一个消费组中，每个分片只会分配给一个消费者
  2. 同一个消费组中，消费者可以被指派多个分片
  3. 不同消费组可以同一时间消费同一个主题

介绍一下再平衡机制？

答：

消费者组再平衡 是一个关键机制，用于管理和分配主题分区给消费者组中的各个消费者。再平衡过程可以确保数据负载在消费者之间均匀分布，并在消费者加入或离开时自动调整分区的分配
我记得是有几个分区策略可以选择的
- 分别是：范围分配、轮询分配、粘性分配、合作粘性分配，其中合作粘性分配和粘性分配一样都是尽可能减少变动，不同点是合作粘性分配下，未受变动的消费者可以继续消费主题

Kafka 什么情况下会 Rebalance？

答：

当 kafka 遇到如下三种情况的时候，kafka 会触发 Rebalance 机制：

新消费者加入：当一个新的消费者加入消费者组时，Kafka 需要重新分配分区，以包括新的消费者
消费者离开：当一个消费者离开（无论是正常关闭还是崩溃）时，需要重新分配该消费者负责的分区给其他消费者
主题分区变化：当主题的分区数量发生变化时（例如：增加新的分区），Kafka 需要重新分配这些分区

Rebalance 有什么影响？

答：

重复消费，如果某个消费者离开消费组时还没来得及提交 offset，当再平衡之后，接盘对应分区的消费者就会重复消费，浪费资源
性能变差，再平衡是需要相对复杂的流程去实施的，在实施再平衡的这个过程中，消费速度也会受到影响

所以我们需要避免频繁再平衡，业务需要导致的变化我们管不了，但我们可以调节心跳相关参数，避免存活误判导致的再平衡

补充：

介绍关注如下参数：

session.timeout.ms： 一次 session 的连接超时时间，也就是说超过这个时间没心跳就会判定消费组超时触发再平衡，如果想避免误判，这个时间可以设置稍微大一些，比如：5s、10s
heartbeat.interval.ms：心跳时间，如果设置过大，就容易触发频繁再平衡，比如：session.timeout.ms 设置为 10s，如果你心跳时间也是设置为 10s，那很容易就超时了，一般都要保证在超时时间之内，至少有 35 次心跳机会，比如：这里可以设置为3s、4s、5s。
max.poll.interval.ms：可以调节这个参数避免活锁问题，Consumer 消费时间过长超过这个参数，会导致 Coordinator 错误地认为 “已停止” 从而被 “踢出” Group

介绍一下重平衡的具体执行流程？

答：

我理解有如下几个步骤
首先是暂停消费，其作用是防止在重新分配期间发生数据丢失或重复，接着由消费组协调器触发再平衡，进而重新分配分区，最后就是开启消费，简单来说就是通知消费者，然后消费者就可以恢复消费了

补充：

有 5 个步骤，依次是：

暂停消费：在再平衡过程中，消费者会暂停对消息的消费，以防止在重新分配期间发生数据丢失或重复
触发再平衡：由消费者组协调器（通常是 Kafka 集群中的一个 Broker）触发再平衡
重新分配分区：协调器根据当前消费者组的成员重新分配主题的分区
通知消费者：重新分配完成后，协调器会通知所有消费者新的分配情况
恢复消费：消费者收到新的分配后，恢复消费，开始处理被分配到的新分区

介绍一下重平衡时的分区策略？

答：

大的方向分两类
- 第一类是 急切再平衡（Eager Rebalance）：Range Assignor、RoundRobin Assignor、Sticky Assignor 都属于 Eager Rebalance，可以理解为急切的再平衡，因为它太过急切，所以顾不得那么多，为了快点搞定这件事，一旦开启再平衡所有消费者都会停止从 Kafka 消费并放弃其分区的成员资格
- 第二类是 增量再平衡（Incremental Rebalance）：CooperativeStickyAssignor 这个 2.3 版本之后引入的优化策略就，属于这一类，在此模式下，只有部分分区会从某个消费者移动到另外一个消费者，其它不受重新平衡影响的 Kafka 消费者可以继续处理数据而不会中断
当然，这样一次执行下来可能分配个数是不均匀的，所以整个消费者组可以经历多次重新平衡，直到找到稳定的分配，因此称为 增量重新平衡，可以理解为 一点一点地来寻求重平衡。相比于急切重平衡，优点在于 消费不会全部暂停，且消费者的分配关系变动较小，当然付出的代价就是完成再平衡的时间可能会更久一些。

补充：

分区策略（也被叫做分区协议）由 partition.assignment.strategy 这个参数配置，回顾一下，它有如下几种选择：
1. Range Assignor：基于范围的分配策略，将分区按照范围分配给消费者
2. RoundRobin Assignor：基于轮询的分配策略，分区均匀地分配给消费者
3. Sticky Assignor：优先保持当前的分配状态，并尽量减少在再平衡过程中的分区移动
4. CooperativestickyAssignor：和 Sticky Assignor 的策略是一样，区别在于未受变动的消费者可以继续消费主题
从再平衡的视角，这几种分区策略大的来说其实可以分为两个 “阵营”，一个叫 Eager Rebalance，一个叫 Incremental Rebalance

解释下 Kafka 中位移（offset）的作用？

如何控制消费的位置？

答：

每条消息在 Kafka 中会有 Partition ID 以及 offset，通过这两个信息就可以定位到一条消息。消费者组消费消息之后会提交它在某个 Partition 对应的 offset，这样子下一次就可以从这个位置开始消费

补充：

Consumers

从上图我们能看到如下几个关键信息:
1. 不同消费者可以在同一时间对同一主题进行消费
2. 相同消费者可以同一时间从同一主题的不同分片读取信息
3. 如果一个消费者，同时消费多个分片下，无法保证消息之间的先后顺序
4. 如果一个消费者，只消费一个分片，消费顺序即生产顺序，符合队列的先入先出特性
默认情况下，Kafka 的消费者只会消费连接之后新产生的消息，如果想消费历史消息，是需要传递参数指定的

Offset

每条消息在 Kafka 中会有 Partition ID 以及 OFFSET，通过这两个信息就可以定位到一条消息。消费者组消费消息之后会提交它在某个 Partition 对应的 OFFSET，这样子下一次就可以从下个位置（OFFSET + 1）开始消费
比如上图，消费了 5 这个位置的消息，也做了对应的提交，所以 Commited offset 也是 5，下一次消费的就是 6 这个位置的消息了
同时，如果一个指定的 offset 被确认，那么它之前的信息就相当于都确认了，下次消费是从它的下一条消息开始消费

Consumer 默认从哪里开始消费？

答：

默认 从提交偏移之后的位置开始消费，如果没有提交过偏移，那么会根据 auto.offset.reset 这个配置决定从哪里开始消费，默认是 latest，消费者将从当前最新的数据开始读取，如果业务需要，还可以选择从最早的偏移开始读取

补充：

auto.offset.reset 是 Apache Kafka 中消费者配置的一部分，用于指定消费者在没有初始偏移量（offset）或当前偏移量在服务器上不存在的情况下应如何处理的策略
- 这个配置参数有三个可能的值：
  1. latest（默认值）：消费者将从最新的偏移量开始读取。这意味着它将开始从消息流的未尾读取数据
  2. earliest：消费者将从最早的偏移量开始读取。这意味着它将从消息流的开头读取所有数据。
  3. none（很少用到）：如果找不到先前的偏移量且没有定义的偏移量，消费者将抛出一个异常。这种情况下，需要确保在启动消费者前，明确定义要消费的偏移量

Consumer 怎么手动指定开始消费偏移？

答：

Java 客户端有方法能指定开始消费的 Offset，函数名应该就是 seek，将希望的偏移量传入这个方法即可

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {

    public static void main(String[] args) {
        // 配置Kafka消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题和分区
        String topic = "test-topic";
        TopicPartition partition = new TopicPartition(topic, 0); // 假设使用分区0
        consumer.assign(Collections.singletonList(partition));

        // 指定偏移量
        // !!!!!!!!
        long offsetToStart = 10L; // 从偏移量10开始消费
        consumer.seek(partition, offsetToStart);

        // 消费消息
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n",
                            record.offset(), record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

Kafka 消费消息是推还是拉？

答：

Kafka 的消费者使用的 拉模式 来获取信息，也就是说每次消费者是发消息到 Kafka 的 Broker 来获取信息，而不是由 Kafka 的 Broker 主动推送
选择拉模式的主要原因还是为了让消费者可以按自身情况来控制消费速度，根据系统资源利用情况（如： CPU、内存等）、业务需要等因素合理拉取消息，避免因消息处理速度不合理带来的资源浪费或过载

Kafka 消费者提交之后就会清理掉数据吗？

答：

在 Kafka 中，如果消息被消费者消费并提交了对应偏移，这条消息不会被删除，可以通过更改该消费者的偏移再次消费，也可以被其它消费者消费（只要消费者消费偏移不超过该消息对应的偏移）

实战经验

Kafka 中什么情况下会出现消息丢失的问题？

答：

消息生产时如果设置的 acks 不是全部副本，那么如果在 follower 副本未完成同步之前，leader 副本挂掉了，消息就会丢失
存储时如果没有用多副本备份，消息也可能会丢失
最后就是消费时，如果没有确认消费成功再提交 offset，而这时候消费者又挂掉了，那么消息同样会丢失

补充：

从消息流转环节分析，分别考虑生产环节、存储环节、消费环节

Kafka 如何保证消息不丢失？

答：

首先 为主题分区配置好多副本，并且 设置写入 acks 参数为全部副本
最后就是 消费时候一定要确认消费成功再提交 offset，这样即使消费者挂掉了，重启之后也能拉到原来那条未成功消费的消息
也可以提一下 发送消息要用带回调的方法，并且重试次数设置得大一些，比如：10以上，避免瞬时抖动带来的失败

补充：

从消息流转环节分析，分别考虑生产环节、存储环节、消费环节

MQ 消息积压了怎么办？

答：

如果 分区数大于消费者数量，那么通过扩容消费端的实例数来提升总体的消费能力；如果相等，那么既需要扩容消费者数量同时需要扩容分区数
如果 短时间内没有足够的服务器资源进行扩容，可以考虑将系统降级，通过关闭一些不重要的分支业务，让系统还能正常运转，服务一些重要业务
还有一种不太常见的情况，你通过监控发现，无论是发送消息的速度还是消费消息的速度和原来都没什么变化，这时候你 需要检查一下你的消费端，是不是消费失败导致的一条消息反复消费这种情况比较多，这种情况也会拖慢整个系统的消费速度。如果监控到消费变慢了，你需要检查你的消费实例，分析一下是什么原因导致消费变慢。优先检查一下日志是否有大量的消费错误，如果没有错误的话，可以通过打印堆栈信息，看一下你的消费线程是不是卡在什么地方不动了，比如触发了死锁或者卡在等待某些资源上了

补充：

三板斧：扩容、降级、排查异常

Kafka 如何保证消息不重复消费？

答：

消费逻辑需要是幂等的，保证不产生重复影响，!实现方式很多，比如：MySQL 设置唯一索引、额外使用记录表来判重等方式

补充：

kafka 出现消息重复消费的原因：
- 消费者宕机、重启或者被强行 kill 进程，导致消费者消费的 offset 没有提交，恢复正常后可能会重复消费
- 由于消费者端处理业务时间长导致会话超时，那么就会触发 reblance 重平衡，此时可能存在消费者 offset 没提交，会导致重平衡后重复消费
要注意重复是不可避免的，重要的是保证不产生重复影响，即：消费端需要保证幂等性

Kafka 如何实现精准一次语义？

答：

本质就是 不重复 + 不丢失

不重复 的核心是 幂等消费
不丢失 的核心是 为主题分区配置好多副本，并且设置写入 acks 参数为全部副本，同时消费时候一定要确认消费成功再提交 offset，这样即使消费者挂掉了，重启之后也能拉到原来那条未成功消费的消息

补充：

至少一次 + 最多一次，不重复 + 不丢失

假设你有个业务希望进入 Kafka 的消息都是有序的，你会怎么做？

答：

Kafka 的分片流入规则是这样的：
- 如果指定了 Partition，那么就是发送到特定的 Partition；如果没有指定 Partition，但是指定了一个 Key，那么就是根据 Key 的 Hash 取模来决定是哪个 Partition；如果都没有指定，就是依次轮替着写入
- 所以我们可以 用一个能标识业务的唯一名字来当 Key，比如：秒杀，就叫 SecKill，指定 Key 之后算出来一定是落在相同的 Partition，也就保证了顺序

高可用

Replica、Leader 和 Follower 三者的概念？

答：

Replica：Replica 是指 Kafka 集群中的一个副本，它可以是 Leader 副本或者 Follower 副本的一种。每个分区都有多个副本，其中 一个是 Leader 副本，其余的是 Follower 副本。每个副本都保存了分区的完整数据，以保证数据的可靠性和高可用性
Leader：Leader 是指 Kafka 集群中的一个分区副本，它负责处理该分区的所有读写请求。Leader 副本是唯一可以向分区写入数据的副本，它将写入的数据同步到所有的 Follower 副本中，以保证数据的可靠性和一致性
Follower：Folower 是指 Kafka 集群中的一个分区副本，Follower 副本不能直接向分区写入数据，它只能从 Leader 副本中复制数据，并将数据同步到本地的副本中，以保证数据的可靠性和一致性。在Leader 副本挂掉的时候，Follower 副本有机会被选举为新的 Leader 副本从而保证分区的可用性

Kafka 中 AR、ISR、OSR 三者的概念？

答：

AR（Assigned Replicas）：AR 是指 分区的所有副本，包括：Leader 副本和 Follower 副本
ISR（In-Sync Replicas）：ISR 是指 与 Leader 副本保持同步的副本集合。ISR 中的副本与 Leader 副本保持同步，即：它们已经复制了 Leader 副本中的所有数据，并且与 Leader 副本之间的数据差异不超过一定的阈值（Follower 副本能够落后 Leader 副本的最长时间间隔）。并且 ISR 副本集合是动态变化的，不是一成不变的。ISR 中的副本可以被选举为新的 Leader 副本，以保证分区的正常运行
OSR（Out-of-Sync Replicas）：OSR 是指 与 Leader 副本不同步的副本集合。OSR 中的副本与 Leader 副本之间的数据差异超过了一定的阈值，或者它们还没有复制 Leader 副本中的所有数据。OSR 中的副本不能被选举为新的 Leader 副本，除非开启了 Unclean 选举。它们只能等待与 Leader 副本同步，或者被替换为新的副本

分区副本什么情况下会从 ISR 中剔出？

答：

每个 Partition 都会由 Leader 动态维护一个与自己基本保持同步的 ISR 列表。所谓动态维护，就是说如果一个 Follower 比一个 Leader 落后超过了给定阈值，默认是 10s，则 Leader 将其从 ISR 中移除。如果OSR 列表内的 Follower 副本与 Leader 副本保持了同步，那么就将其添加到 ISR 列表当中

分区副本中的 Leader 如果宕机但 ISR 却为空该如何处理？

答：

可以通过 unclean 选举配置参数（即：unclean.leader.election，通常不用说这么长的配置名字）来决定是否从OSR中选举出 Leader：
- 如果是该参数是 true：允许 OSR 成为 Leader，但是 OSR 的消息较为滞后，可能会出现消息丢失的问题
- 否则：坚决不能让那些 OSR 竞选 Leader。这样做的后果是这个分区就不可用了

分区副本之间同步，是推还是拉（kafka如何主从同步）？

答：

数据是先写入到 Leader 副本，同步时候是 Follower 副本去主动拉取消息，拉的优势在于副本机器可以根据自身的负载情况来拉取

高可用机制是怎么实现的？

答：

Kafka 天然支持 多副本机制，每个副本都有完整的数据，这些副本分散在不同的 Broker 上，就算主副本所在Broker 的磁盘损坏了，其它 Broker 也能把数据找回来并升级为主副本，通过这种方式 Kafka 就实现了高可用

补充：

Kafka 受欢迎有一个很大的原因是它天然提供了容灾解决方案，可以应对机器故障等各种异常，这些异常我们很多是无法预防的，比如：机房断电，机器硬盘损坏，甚至之前出现过的天津机房爆炸事故
Kafka 高可用机制主要就体现在多副本上，说直白点，副本横跨多个 Broker / 机器，一个副本挂了还有另一个这里回答的要点包括：
1. 多副本，相当于数据有了多个备份
2. 多副本是横跨多 Broker 的，这样在机器级别也有了容灾能力

Kafka 是怎么为分片选择主副本的？

答：

Kafka 维护了一个叫 ISR 的列表，ISR 里的副本都是包含完整数据的，当没有 Leader，或原有 Leader 挂掉了，Kafka 就会从 ISR 列表中选择第一个副本升级为 Leader

补充：

关键知识点是 ISR，一个分区会有多个副本，为了实现更好的管理，可以将 Kafka 中的副本划分成不同的集合：
1. AR（Assigned Replicas）：AR 是指 分区的所有副本，包括：Leader 副本和 Follower 副本
2. ISR（In-Sync Replicas）：ISR 是指 与 Leader 副本保持同步的副本集合。ISR 中的副本与 Leader 副本保持同步，即：它们已经复制了 Leader 副本中的所有数据，并且与 Leader 副本之间的数据差异不超过一定的阈值（Follower 副本能够落后 Leader 副本的最长时间间隔）。并且 ISR 副本集合是动态变化的，不是一成不变的。ISR 中的副本可以被选举为新的 Leader 副本，以保证分区的正常运行
3. OSR（Out-of-Sync Replicas）：OSR 是指 与 Leader 副本不同步的副本集合。OSR 中的副本与 Leader 副本之间的数据差异超过了一定的阈值，或者它们还没有复制 Leader 副本中的所有数据。OSR 中的副本不能被选举为新的 Leader 副本，除非开启了 Unclean 选举。它们只能等待与 Leader 副本同步，或者被替换为新的副本
简单来说，OSR 集合 = AR 集合 - ISR 集合

Kafka 怎么知道 Leader 挂了？

答：

Broker 中会选出一个来担任 Controller，选 Controller 负责监测 Leader 的状态，这样 Leader 挂掉之后 Controller 就能感知到，并从 ISR 选择出新的 Leader
- 所有的 Broker 会尝试在 Zookeeper 中创建数据 / controller，先创建成功的就是 controller
- 注意，这是个临时数据，如果当前的 Controller 挂掉或者网络掉线，这个数据就会消失。其它 Broker 通过 Watch 感知到这件事之后，就会再次抢写

补充：

Kafka 比较早期就直接让分片副本抢写 Zookeeper，谁先写入谁就是 Leader，但是这样有个潜在问题：当分区和副本数量比较多的时候，所有的副本都直接参与选举，对 ZooKeeper 的压力会比较大
更新一些的版本，Kafka 做了优化，与其让所有副本都来卷，不然就先选举一个 Broker 作为 Controller，也就是控制者
Controller 职责：感知 Broker、Topic、Partition 的变化，维护 Partition 的 Leader 信息
有了 Controller再结合 ISR 机制，就可以直接指定 ISR 集合中的第一个备选 Broker 作为新 Leader
由 Controller 来持续监测 Leader，如果发现 Leader 挂掉了再由 Controller 从 ISR 中选择新的 Leader

高性能

Kafka 为什么这么快？

Kafka 性能为什么这么高？

Kafka 吞吐量为什么这么大？

答：

我知道有几个点都有提升 Kafka 处理的速度，包括：顺序写、零拷贝、数据压缩、批量操作（说自己比较熟悉的，多一个少一个，无所谓），其中我觉得影响最大或者说最具 Kafka 特色的，就是顺序写，通过磁盘的顺序写入，非常直接优化了兼顾了性能和复杂度，其次我印象最深的是批量操作和数据压缩，这两个也是业务优化的常见思路

聊聊 Kafka 顺序写机制？

答：

顺序写指的是按顺序将数据写入磁盘，对于 Kafka 而言，其实就是直接追加到磁盘文件末尾，顺序写性能非常高，甚至接近内存写，所以出于 Kafka 自身定位、复杂度、性能等综合考虑，Kafka 选择了顺序写

补充：

要点：

Kafka 数据写入就是直接追加到磁盘文件未尾，这就是顺序写
顺序写性能非常高，甚至接近内存写
Kafka 从复杂度和性能综合考虑，选择了顺序写

聊聊 Kafka 页缓存机制？

答：

Page Cache 可以简单看作 热点磁盘数据的内存缓存，当消息写入时，是先写入 Page Cache，后面由操作系统将其刷入磁盘，这样性能就会提升很多，同时，如果查询时候发现 Page Cache 中有对应数据，那么也就不用去磁盘读取
值得一提的是，Kafka 是生产消费者模式，即：生产了消息，在无积压情况下，这个消息很快就会被消费，也就是说我们生产消费时写入了 Page Cache，而很快就有消费者来触发 Kafka 应用程序读取对应数据，而这个时间间隔很短，Page Cache 命中的可能性会很高，自然提升效果就会非常大了

补充：

Page Cache 是提升 Kafka 性能的重要机制，通过合理配置 Kafka 参数和操作系统参数，可以充分利用 Page Cache 提高消息的读写性能。了解 Page Cache 的工作原理并进行适当的优化，对于保障 Kafka 集群的高效运行至关重要。
要点：

要说到是操作系统特性
要解释什么地方用到了 Page Cache
要说明为何对读写优化效果

聊聊 Kafka 零拷贝机制？

答：

零拷贝技术可以优化数据发送效率，本质上是用 sendfile 系统调用，取代了 write 和 read 系统调用，让系统调用次数从 2 次变成了一次，相关内核上下文切换次数从 4 次变成了 2 次，同时配合 DMA 技术，让数据拷贝从 4 次变成了 2 次

补充：

零拷贝技术可以优化数据发送效率，这是 Linux 本身的特性，而 Kafka 则是这个特性的使用者，通过最终调用 sendfile 系统调用来享受了零拷贝的福利
回答要点：

减少系统调用
减少内核上下文切换
减少拷贝次数

Kafka之前的拷贝机制

Kafka采用的零拷贝机制

聊聊 Kafka 分层设计机制？

答：

分层设计其实就是出自分治思考，对于 Kafka 而言，所有消息写入一个文件，那肯定扛不住，所以提出了 Topic 的概念，可以一个业务写入一个 Topic，单个 Topic 不具备扩展性，扛不住大流量业务，所以 Topic 又进行了分片，也就是 Partition，一个业务的消息可以根据规则写入多个 Partition，单个 Partition 是不是也需要能扩展，于是一个 Partition 又可以切分为多个 segment 文件（根据 offset 排列），segment 文件支持按需滚动增长，所以 Partition 就具备了扩展性。这就是 Kafka 一以贯之的分层设计

补充：

要点：

可以提一下分治思想
Topic、Partition、Segment 这些概念可以连通讲一遍
抓住扩展性来聊

Kafka 文件高效存储设计原理？

答：

Kafka 把 Topic 中一个大的 Parition 文件分成多个小的 segment file，通过多个 segment file，就容易定期清除或删除已经消费完的文件，减少磁盘的占用
通过 索引元数据 来管理消息的位置和偏移量，以便快速定位和读取消息
通过 索引元数据 全部映射到内存，可以避免 segment file 的 I/O 磁盘操作
通过 索引文件 稀疏存储，可以大幅降低索引文件元数据占用空间大小

聊聊 Kafka 哪些环节用了批量操作？

答：

Kafka 主要有2个批量操作的地方：
- 一个是 批量生产，也就是批量发送，其实就是通过发送缓冲，将数据缓冲起来，等聚集了一批数据，再一次性发送给 Broker
- 另一个是 批量消费，本质就是一次多拉几条消息，一起消费

聊聊 Kafka 数据压缩？

答：

通过压缩可以让传输的数据变小，以节约带宽，所以我们可以通过压缩消息来提升 Kafka 性能，一般而言就是在 发送方 进行压缩，有时候也可以 在 Broker 侧 进行压缩。压缩 适用于 CPU 比较富裕，带宽相对不足的情况，而消息的传输大多数是符合这个情况的

补充：

理解压缩其实就是让数据“变小”，这样效率自然就上去了。可以聊聊在哪里压缩，什么情况下适合压缩

其他

Zookeeper 对于 Kafka 的作用是什么？

答：

Zookeeper 中存储了 Broker 的信息，Broker 会向 Zookeeper 发送 心跳请求 来上报自己的状态
Zookeeper 中存储了所有的 Topic 的注册信息，包括：Topic 列表，每个 Topic 的 Partition 数量，副本在哪一个 Broker（ISR）等等
控制器选举：kafka 集群有多个 broker，其中一个会被选举为控制器保存在 Zookeeper 中（在 /controller 中维护），控制器负责 brokers 的上下线，管理整个集群所有分区和副本的状态，如果某个分区的 Leader 故障了，控制器会选举出新的 Leader
同时 Zookeeper 也用来记录一些配置信息，比如：主题的配置信息，这样将配置信息存储在 Zookeeper 这种第三方组件，所有的 Broker 都可以很方便地拿到对应配置以及感知到相应变化
分区注册：Kafka 的每个 partition 只能被某个消费者组的同一个 consumer 消费，kafka 必须知道所有的 partition 与 consumer 的关系，因此消费者与分区的关系被保存在 Zookeeper 当中

补充：

Zookeeper 拥有分布式协调能力，Kafka 主要是用 Zookeeper 来管理 Broker / Topic 数据、存储配置、选择 Controller，曾经也会存储消费者偏移信息，但是因为 Zookeeper 不适合做频繁交互的操作，所以后面砍掉了记录消费提交这点，尽可能让 Zookeeper 的压力更轻，专注于协调相关事情

Kafka 你用的是哪个版本？

答：

我们实验室在项目中用的是 Kafka 2.5.0 版本，主要原因是这个版本团队用了很多年都是非常稳定的所以新老项目都一直延用这个版本