kafka原理2-知识点

• kafka使用tcp, io多路复用即信号驱动io, 生产者有同步和异步2种类型

• 选择由producer向broker push消息并由consumer从broker pull消息。

• kafka和nsq对比：

  • kafka是pull，有序，吞吐高；nsq消息是无序的，吞吐低，有requeue和defer功能，不持久化，不可回溯，pull，用内存，所以速度快。
• kafka数据可靠性和重复消费
  1. 需要消费者操作幂等，来保证重复消费无影响
  2. 处理后提交commit，保证消息被消费到，事务保证
  3. 生产者生产消息失败时，报error。
  4. 如果要保证有序，让消息到1个partition就行了，partition内部消费是有序的

• kafka基于zk.

• kafka是发布-订阅模型。

• Zookeeper 主要为 Kafka 做了下面这些事情：

  1. Broker 注册 ：在 Zookeeper 上会有一个专门用来进行 Broker 服务器列表记录的节点。每个 Broker 在启动时，都会到 Zookeeper 上进行注册，即到/brokers/ids 下创建属于自己的节点。每个 Broker 就会将自己的 IP 地址和端口等信息记录到该节点中去
  2. Topic 注册 ： 在 Kafka 中，同一个Topic 的消息会被分成多个分区并将其分布在多个 Broker 上，这些分区信息及与 Broker 的对应关系也都是由 Zookeeper 在维护。比如我创建了一个名字为 my-topic 的主题并且它有两个分区，对应到 zookeeper 中会创建这些文件夹：/brokers/topics/my-topic/Partitions/0、/brokers/topics/my-topic/Partitions/1
  3. 负载均衡 ：上面也说过了 Kafka 通过给特定 Topic 指定多个 Partition, 而各个 Partition 可以分布在不同的 Broker 上, 这样便能提供比较好的并发能力。 对于同一个 Topic 的不同 Partition，Kafka 会尽力将这些 Partition 分布到不同的 Broker 服务器上。当生产者产生消息后也会尽量投递到不同 Broker 的 Partition 里面。当 Consumer 消费的时候，Zookeeper 可以根据当前的 Partition 数量以及 Consumer 数量来实现动态负载均衡。
  4. ……

• ISR:In-Sync Replicas 副本同步队列 AR:Assigned Replicas 所有副本

  HW:High Watermark 高水位，取一个partition对应的ISR中最小的LEO作为HW，consumer最多只能消费到HW所在的位置上一条信息。

  LEO:LogEndOffset 当前日志文件中下一条待写信息的offset

  HW/LEO这两个都是指最后一条的下一条的位置而不是指最后一条的位置。

  LSO:Last Stable Offset 对未完成的事务而言，LSO 的值等于事务中第一条消息的位置(firstUnstableOffset)，对已完成的事务而言，它的值同 HW 相同

  LW:Low Watermark 低水位, 代表 AR 集合中最小的 logStartOffset 值

支持语义和exactly-once实现

支持语义：

• 3种都支持。
• 在 0.10 之前并不能保证 exactly-once，需要使用 Consumer 自带的幂等性保证。0.11.0 使用事务保证了。实现上是broker去重

Kafka 分别通过 幂等性（Idempotence）和事务（Transaction）这两种机制实现了 精确一次（exactly once）语义。

at least once

• ack = all，同步多写到broker
• 消费时手动提交offset

Exactly once

• 事务型Producer，同时开启幂等性producer
• 事务型Consumer

消息发送幂等性

Kafka 的生产者幂等性是通过引入 producer id (PID) 和 序列号 (sequence number) 来实现的，

通过 Producer ID 和 序列号，Kafka 可以确保生产者发送的消息不会因重复、乱序等原因影响消费者的消费，保证了消息的幂等性。生产者在发送消息时，通过维护每个分区的序列号，避免了消息丢失或重复消费的问题。

实际上是在broker上检查是否有递增序号丢失，丢了就producer提交时报错。然后去重保证消息不重复。

注意同样的限制也比较大：

• 首先，它只能保证单分区上的幂等性。即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息，它无法实现多个分区的幂等性。

  • 因为 SequenceNumber 是以 Topic + Partition 为单位单调递增的，如果一条消息被发送到了多个分区必然会分配到不同的 SequenceNumber ,导致重复问题。

• 其次，它只能实现单会话上的幂等性。不能实现跨会话的幂等性。当你重启 Producer 进程之后，这种幂等性保证就丧失了。

  • 重启 Producer 后会分配一个新的 ProducerID，相当于之前保存的 SequenceNumber 就丢失了。

生产者幂等性实现的关键点：

1. Producer ID (PID)： 每个新的生产者实例在初始化时会被分配一个唯一的 PID。这个 PID 是透明的，用户无需关心。
2. 序列号 (Sequence Number, SN)： 每个生产者实例的每个分区都有一个独立的序列号，序列号从 0 开始，并且每发送一条消息，序列号递增。
3. Broker 端序列号管理： 在 Kafka broker 端，会为每一对 <PID, 分区> 维护一个序列号，这个序列号记录了生产者向该分区发送的最后一条消息的序列号。

消息接收与处理：

• 消息接收条件

  对于每条新到的消息，broker 会检查该消息的序列号SN_new 和 当前分区对应的序列号 SN_old

  • SN_new = SN_old + 1：表示消息按序列号顺序发送，broker 接收该消息并更新序列号。
  • SN_new < SN_old + 1：表示消息是重复发送的，broker 会丢弃该消息。
  • SN_new > SN_old + 1：表示消息乱序，可能发生了消息丢失，生产者会抛出 OutOfOrderSequenceException 异常，提示数据丢失。

异常处理：

• 如果发生 OutOfOrderSequenceException 异常，说明消息的发送顺序出现问题，生产者会收到这个异常，并且后续的操作（如 send()、beginTransaction()、commitTransaction() 等）会抛出 IllegalStateException 异常。

事务

• KAFKA 的事务机制，是 KAFKA 实现端到端有且仅有一次语义（end-to-end EOS)的基础；
• KAFKA 的事务机制，涉及到 transactional producer 和 transactional consumer, 两者配合使用，才能实现端到端有且仅有一次的语义（end-to-end EOS)；
• 当然kakfa 的 producer 和 consumer 是解耦的，你也可以使用非 transactional 的 consumer 来消费 transactional producer 生产的消息，但此时就丢失了事务 ACID 的支持；
• 通过事务机制，KAFKA 可以实现对多个 topic 的多个 partition 的原子性的写入，即处于同一个事务内的所有消息，不管最终需要落地到哪个 topic 的哪个 partition, 最终结果都是要么全部写成功，要么全部写失败（Atomic multi-partition writes）；
• KAFKA的事务机制，在底层依赖于幂等生产者，幂等生产者是 kafka 事务的必要不充分条件；
• 事实上，开启 kafka事务时，kafka 会自动开启幂等生产者。

一个大致的实现流程基本如下：首先向协调者获取事务ID（后文统称TID），然后向参与者发送请求准备提交，带上这个TID，参与者现在本地做append，如果成功返回，协调者持久化决策的内容，然后执行决策，参与者将消息真正写到Log中（更新LSO，与HW高水位区分）。但是上文也讲了2PC实际上是有一些问题的，首先2PC协调者的单点问题，Kafka的解决方法也比较简单，直接利用自己单分区同步复制保证线性一致性的特性，将协调者的状态存储在内部Topic中，然后当协调者崩溃时可以立刻做转移然后根据Topic做恢复，因为Topic本身就单分区而言就是个线性存储。

另外，就是2PC的协调者本质是个主从复制的过程，由于TransactionCoordinator本来就挂靠在Broker上，所以这个选举依然会委托给Controller，这样就解决了2PC中的比较棘手的问题。而对于事务的隔离级别，Kafka仅实现到了“读已提交（RC）”级别。

事务协调者的选举与 Kafka 集群中 Leader 选举 的机制类似，也是基于 ZooKeeper 来实现的。

事务型 Producer 也不惧进程的重启。Producer 重启回来后，Kafka 依然保证它们发送消息的精确一次处理。这个和幂等性不一样。

设置事务型 Producer 的方法也很简单，满足两个要求即可：

• 和幂等性 Producer 一样，开启 enable.idempotence = true。

• 设置 Producer 端参数 transactional. id。最好为其设置一个有意义的名字。在 Kafka 的事务中，应用程序必须提供一个唯一的事务 ID，即 Transaction ID，并且宕机重启之后，也不会发生改变。

Transactin ID 与 PID 可能一一对应，区别在于 Transaction ID 由用户提供，而 PID 是内部的实现对用户透明。

为了 Producer 重启之后，旧的 Producer 具有相同的 Transaction ID 失效，每次 Producer 通过 Transaction ID 拿到 PID 的同时，还会获取一个单调递增的 Epoch。

由于旧的 Producer 的 Epoch 比新 Producer 的 Epoch 小，Kafka 可以很容易识别出该 Producer 是老的，Producer 并拒绝其请求。

为了实现这一点，Kafka 0.11.0.0 引入了一个服务器端的模块，名为 Transaction Coordinator，用于管理 Producer 发送的消息的事务性。

该 Transaction Coordinator 维护 Transaction Log，该 Log 存于一个内部的 Topic 内。

由于 Topic 数据具有持久性，因此事务的状态也具有持久性。Producer 并不直接读写 Transaction Log，它与 Transaction Coordinator 通信，然后由 Transaction Coordinator 将该事务的状态插入相应的 Transaction Log。

Transaction Log 的设计与 Offset Log 用于保存 Consumer 的 Offset 类似。

事务应该看这个。 link 和 这个

此外，你还需要在 Producer 代码中做一些调整，如这段代码所示：

producer.initTransactions();
try {
            producer.beginTransaction();
            producer.send(record1);
            producer.send(record2);
            producer.commitTransaction();
} catch (KafkaException e) {
            producer.abortTransaction();
}

和普通 Producer 代码相比，事务型 Producer 的显著特点是调用了一些事务 API，如 initTransaction、beginTransaction、commitTransaction 和 abortTransaction，它们分别对应事务的初始化、事务开始、事务提交以及事务终止。

这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka，要么它们全部提交成功，要么全部写入失败。

实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。

因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更的，来实现只读已提交的内容。设置 isolation.level 参数的值即可。当前这个参数有两个取值：

• read_uncommitted：这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。
• read_committed：表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。

事务协调器（Transaction Coordinator）是 Kafka 中负责管理和协调事务提交、回滚以及状态管理的关键组件。它的主要任务是确保在分布式环境中，涉及多个分区或多个消息的事务要么成功提交，要么完全回滚，从而保证事务的 原子性 和 一致性。

事务协调器的工作可以分为两个阶段，类似于经典的 两阶段提交（2PC）协议：

1. 阶段 1：准备阶段（Prepare）

• 生产者向 Kafka 发送事务消息，并标记事务为“进行中”。
• Kafka 集群中的各个分区的副本接收到这些消息后，事务协调器将向所有参与的分区确认它们已经接收到消息并准备好提交。
• 如果所有分区都准备好提交事务，协调器将进入第二阶段。如果任何分区无法准备好提交（例如出现了硬件故障），协调器会回滚事务。

1. 阶段 2：提交或回滚（Commit or Abort）

• 如果所有分区都成功接收到消息并准备提交，协调器将向所有分区发出提交事务的请求，并将事务状态更新为“已提交”。
• 如果任何分区失败或事务被中止，协调器会向所有分区发出回滚事务的请求，并将事务状态更新为“已回滚”。

幂等性 Producer 和事务型 Producer 都是 Kafka 社区力图为 Kafka 实现精确一次处理语义所提供的工具，只是它们的作用范围是不同的。

• 幂等性 Producer 只能保证单分区、单会话上的消息幂等性；
• 而事务能够保证跨分区、跨会话间的幂等性。

从交付语义上来看，自然是事务型 Producer 能做的更多。天下没有免费的午餐。比起幂等性 Producer，事务型 Producer 的性能要更差，在实际使用过程中，我们需要仔细评估引入事务的开销，切不可无脑地启用事务。

消息丢失

日志如何保证一致性

在 Kafka 中，保证日志一致性主要依赖于其 副本机制、Leader Epoch 和 ISR (In-Sync Replicas) 等机制。当出现日志不一致的情况时，Kafka 通过一系列的协调机制来确保数据的一致性和可靠性。

Leader Epoch 是为了应对 Kafka 中 Leader 变更 引起的日志不一致问题而设计的。每当 Kafka 分区的 Leader 变更时，Leader Epoch 会递增，从而标记当前 Leader 的日志版本。

日志不一致场景， **根本原因是follower 的 HW 是不可靠的。 ** 参考

• 场景一: follower的HW是不可靠的，follower的HW更新会落后leader的HW更新。可能会宕机后再选举导致新leader缺少已提交的日志。
• 场景二: 假如几个ISR依次全宕机了，然后依次恢复，由于page cache没刷盘，重启会数据丢失。全崩溃的情况下选出的 leader 并不一定包含所有已提交的日志

其实，如果熟悉raft的话，应该已经发现上面分析的场景和raft中的日志恢复很类似，raft中的follower是可能和leader的日志不一致的，这个时候会以leader的日志为准进行日志恢复。而raft中的日志恢复很重要的一点是follower根据leader任期号进行日志比对，快速进行日志恢复，follower需要判断新旧leader的日志，以最新leader的数据为准。

这里的leader epoch和raft中的任期号的概念很类似，每次重新选择leader的时候，用一个严格单调递增的id来标志，可以让所有follower意识到leader的变化。而follower也不再以HW为准，每次奔溃重启后都需要去leader那边确认下当前leader的日志是从哪个offset开始的。

解决日志不一致，但无法解决page cache的问题：

• Leader Epoch 通过版本控制日志，当副本 A 和 B 在同一分区时，如果 A 变成新的 Leader，它会检查自己持有的日志是否与 B 的日志一致。如果日志不一致，A 会确保删除不一致的部分，保持数据一致性。
• 当副本恢复时，它会通过 OffsetsForLeaderEpochRequest 请求获取当前 Leader 的最新日志信息和偏移量，避免恢复过程中数据被截断或丢失。

消息丢失的场景

1. 自动提交 设置offset为自动定时提交，当offset被自动定时提交时，数据还在内存中未处理，此时刚好把线程kill掉，那么offset已经提交，但是数据未处理，导致这部分内存中的数据丢失。

2. 生产者发送消息 发送消息设置的是fire-and-forget（发后即忘），它只管往 Kafka 中发送消息而并不关心消息是否正确到达。不过在某些时候（比如发生不可重试异常时）会造成消息的丢失。这种发送方式的性能最高，可靠性也最差。

3. 消费者端 先提交位移，但是消息还没消费完就宕机了，造成了消息没有被消费。自动位移提交同理

4. broker消息丢失，比如acks没有设置为all。 如果在broker还没把消息同步到其他broker的时候宕机了，那么消息将会丢失。

Broker即使设置acks=-1，并且选出isr中的副本作为leader的时候，仍然是会存在丢数据的情况的：

broker端极端仍丢失消息的情况

broker端接收到生产端的消息后，并成功应答生产端后，消息会丢吗？ 如果broker能像mysql服务器一样，在成功应答给客户端前，能把消息写入到了磁盘进行持久化，并且在宕机断电后，有恢复机制，那么我们能说broker端不会丢消息。

但broker端提供数据不丢的保障和mysql是不一样的。broker端在接受了一批消息数据后，是不会马上写入磁盘的，而是先写入到page cache里，这个page cache是操作系统的页缓存（也就是另外一个内存，只是由操作系统管理，不属于JVM管理的内存），通过定时或者定量的的方式（ log.flush.interval.messages和log.flush.interval.ms）会把page cache里的数据写入到磁盘里。

如果page cache在持久化到磁盘前，broker进程宕机了，这个时候不会丢失消息，重启broker即可；如果此时操作系统宕机或者物理机宕机了，page cache里的数据还没有持久化到磁盘里，此种情况数据就丢了。

kafka应对此种情况，建议是通过多副本机制来解决的，核心思想也挺简单的：如果数据保存在一台机器上你觉得可靠性不够，那么我就把相同的数据保存到多台机器上，某台机器宕机了可以由其它机器提供相同的服务和数据。

要想达到上面效果，有三个关键参数需要配置 第一：生产端参数 ack 设置为all

第二：在broker端 配置 min.insync.replicas参数设置至少为2。 这个是认为同步了多少才算ack成功。

第三：在broker端配置 replicator.factor参数至少3

以及 配置适当的刷盘策略

总结

Kafka没有提供同步刷盘的方式。同步刷盘在RocketMQ中有实现，实现原理是将异步刷盘的流程进行阻塞，等待响应，类似ajax的callback或者是java的future。

面试题

• kafka面试题
• 整理，答案不全的
• 面试题3
• github面试题
• 面试题总结

QA

有哪些情形会造成重复消费？

1. Rebalance 一个consumer正在消费一个分区的一条消息，还没有消费完，发生了rebalance(加入了一个consumer)，从而导致这条消息没有消费成功，rebalance后，另一个consumer又把这条消息消费一遍。
2. 消费者端手动提交 如果先消费消息，再更新offset位置，导致消息重复消费。
3. 消费者端自动提交 设置offset为自动提交，关闭kafka时，如果在close之前，调用 consumer.unsubscribe() 则有可能部分offset没提交，下次重启会重复消费。
4. 生产者端 生产者因为业务问题导致的宕机，在重启之后可能数据会重发

kafka不用Quorum原因

Kafka的副本策略称为ISRs(in-sync replicas)，动态维护了一个包含所有已提交日志的节点集合，通过zookeeper存储该集合，并由zookeeper从集合中选出一个节点作为leader，日志会先写入到leader，再由ISRs中的其他follower节点主动进行复制同步。

关于为什么Kafka没有采用类似Raft、Paxos的Quorum算法，官方文档也做了说明：

对于大吞吐日志系统而言，用Quorum太浪费。如果要容忍f个节点失败，Quorum需要2 * f + 1个节点，而ISRs只需要f+1个。

我们知道Quorum算法是为了解决脑裂问题，而ISRs这里不会出现这个问题的原因是zookeeper本身是一个分布式协调服务，可以通过zookeeper保证leader的唯一性。

kafka副本策略另外一个设计是每次日志写入并不会进行fsync等刷盘操作，刷盘会导致两到三倍的性能损失。崩溃的节点恢复后并不一定拥有完整的数据，但是可以通过和leader重新同步来加入ISRs。

创建topic时如何选择合适的分区数？

可以使用Kafka 本身提供的用于生产者性能测试的 kafka-producer- perf-test.sh 和用于消费者性能测试的 kafka-consumer-perf-test.sh来进行测试。

增加合适的分区数可以在一定程度上提升整体吞吐量，但超过对应的阈值之后吞吐量不升反降。如果应用对吞吐量有一定程度上的要求，则建议在投入生产环境之前对同款硬件资源做一个完备的吞吐量相关的测试，以找到合适的分区数阈值区间。

分区数的多少还会影响系统的可用性。如果分区数非常多，如果集群中的某个 broker 节点宕机，那么就会有大量的分区需要同时进行 leader 角色切换，这个切换的过程会耗费一笔可观的时间，并且在这个时间窗口内这些分区也会变得不可用。

分区数越多也会让 Kafka 的正常启动和关闭的耗时变得越长，与此同时，主题的分区数越多不仅会增加日志清理的耗时，而且在被删除时也会耗费更多的时间。

分区只能增加，不能删除，因此可以先单个分区，然后压测看大概吞吐量，之后再扩分区数量。

聊一聊你对Kafka的Log Compaction的理解

日志压缩（Log Compaction）：针对每个消息的 key 进行整合，对于有相同 key 的不同 value 值，只保留最后一个版本。

这个key是业务方自己指定的

如果应用只关心 key 对应的最新 value 值，则可以开启 Kafka 的日志清理功能，Kafka 会定期将相同 key 的消息进行合并，只保留最新的 value 值。

Kafka目前有哪些内部topic，它们都有什么特征？各自的作用又是什么？

consumer_offsets：作用是保存 Kafka 消费者的位移信息。

• 位移主题的 Key 中应该保存 3 部分内容：gourpId, topic, partitionId
• 可以认为他们是一个kv数据结构。key 是每个消息都有的，partition就是按key进行路由到对应分片的。

transaction_state：用来存储事务日志消息

聊一聊你对Kafka底层存储的理解

页缓存

页缓存是操作系统实现的一种主要的磁盘缓存，以此用来减少对磁盘 I/O 的操作。具体来说，就是把磁盘中的数据缓存到内存中，把对磁盘的访问变为对内存的访问。

使用文件系统并依赖于页缓存的做法明显要优于维护一个进程内缓存或其他结构，至少我们可以省去了一份进程内部的缓存消耗，同时还可以通过结构紧凑的字节码来替代使用对象的方式以节省更多的空间。

此外，即使 Kafka 服务重启，页缓存还是会保持有效，然而进程内的缓存却需要重建。这样也极大地简化了代码逻辑，因为维护页缓存和文件之间的一致性交由操作系统来负责，这样会比进程内维护更加安全有效。

零拷贝

除了消息顺序追加、页缓存等技术，Kafka 还使用零拷贝（Zero-Copy）技术来进一步提升性能。所谓的零拷贝是指将数据直接从磁盘文件复制到网卡设备中，而不需要经由应用程序之手。零拷贝大大提高了应用程序的性能，减少了内核和用户模式之间的上下文切换。对 Linux 操作系统而言，零拷贝技术依赖于底层的 sendfile() 方法实现。对应于 Java 语言，FileChannal.transferTo() 方法的底层实现就是 sendfile() 方法。

聊一聊Kafka的延时操作的原理

Kafka 中有多种延时操作，比如延时生产，还有延时拉取（DelayedFetch）、延时数据删除（DelayedDeleteRecords）等。 延时操作创建之后会被加入延时操作管理器（DelayedOperationPurgatory）来做专门的处理。延时操作有可能会超时，每个延时操作管理器都会配备一个定时器（SystemTimer）来做超时管理，定时器的底层就是采用时间轮（TimingWheel）实现的。

在发送延时消息的时候并不是先投递到要发送的真实主题（real_topic）中，而是先投递到一些 Kafka 内部的主题（delay_topic）中，这些内部主题对用户不可见，然后通过一个自定义的服务拉取这些内部主题中的消息，并将满足条件的消息再投递到要发送的真实的主题中，消费者所订阅的还是真实的主题。

发送到内部主题中的消息会被一个独立的 DelayService 进程消费，这个 DelayService 进程和 Kafka broker 进程以一对一的配比进行同机部署。

Kafka 本身并不直接支持延迟消息（delayed messages）

Kafka的那些设计让它有如此高的性能？

Kafka 高性能的核心设计包括：

• 分区与并行消费：提高了系统的扩展性和消费并行性。
• producer可以批量发送：减少了网络传输开销，提高了吞吐量。
• 文件顺序读写与零拷贝， page cache的使用
• 高效的日志存储和索引机制：降低了存储和检索的开销，提升了查询效率。

consumer重启如何避免rebalance

• 是心跳检测是否consumer还在消费的。 Kafka 需要等待 session.timeout.ms 过期才检测到 Consumer 失效。因此调这个参数可控制。
• 配置静态id, 重启后依然认为是老的consumer就不会rebalance。 适用于 Kafka 2.3+，配置 group.instance.id，让 Kafka 认为 Consumer 是固定的，即使短暂掉线也不会触发重平衡。
• Kafka 3.2+ 支持渐进式重平衡，仅调整新增或减少的 Consumer，避免影响整个消费者组。每次rebalance只影响部分。