• 👏作者简介：大家好，我是爱吃芝士的土豆倪，24届校招生Java选手，很高兴认识大家
• 📕系列专栏：Spring源码、JUC源码、Kafka原理
• 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
• 🍂博主正在努力完成2023计划中：源码溯源，一探究竟
• 📝联系方式：nhs19990716，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

消息的精准消费

在前面学到的手动提交位移的时机选择的时候

• 数据处理完成之前先提交偏移量

可能会发生漏处理的现象（数据丢失）

反过来说，这种方式实现了： at most once 的数据处理（传递）语义

• 数据处理完成之后再提交偏移量

可能会发生重复处理的现象（数据重复）

反过来说，这种方式实现了： at least once 的数据处理（传递）语义

当然，数据处理（传递）的理想语义是： exactly once（精确一次）

Kafka 也能做到 exactly once（基于 kafka 的事务机制）

实现中，可以记录为 消息存储在一张表，然后偏移量存储在一张表，但是还是有可能出现问题，除非绑定为原子操作。

相当于 偏移量的更新 和 业务数据的落地绑定成一个事务

begin transactioninsert into tb1 values();
insert into t_offset values();commit

还有一种办法就是 利用幂等性，重复就重复，但是插入数据库中的机会就只有一次，那么就能达到最终一致的效果。

所以解决数据重复的问题，有两种解决办法：

1.利用事务

2.利用幂等性

而解决数据丢失的问题，主要有三种解决办法：

1. 启用Kafka的事务机制：Kafka提供了事务机制，可以将消息的处理和偏移量的提交放在同一个事务中进行，确保消息的处理和偏移量的提交是原子性的。通过事务机制，可以避免在数据处理完成之前就提交偏移量而导致数据丢失的问题。
2. 手动控制偏移量的提交：可以在应用程序中手动控制偏移量的提交时机。例如，可以在消息处理完成并且已经被确认成功后再提交偏移量。这样可以确保消息得到正确处理后再进行偏移量的提交，避免数据丢失的问题。
3. 使用At-Least-Once语义：在消费者的配置中设置enable.auto.commit为false，然后手动提交偏移量。在消息处理过程中，确保消息处理的幂等性，即多次处理同一条消息的结果是一致的。这样即使存在重复消息的情况，也能保证数据最终被处理一次。

public class Consumer实现ExactlyOnce手段1{public static void main(String[] args){// 定义 kakfa 服务的地址，不需要将所有 broker 指定上props.put("bootstrap.servers", "doitedu01:9092");// 制定 consumer groupprops.put("group.id", "g1");// key 的反序列化类props.put("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");// value 的反序列化类props.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");// 是否自动提交 offset props.put("enable.auto.commit", "false");// 如果没有消费偏移量记录，则自动重设为起始 offset：latest, earliest, noneprops.put("auto.offset.reset","earliest");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 创建一个jdbc连接Connection conn = DriverManager.getConnection("jdbc:mysql://test:3306/abc","root","123456");conn.setAutoCommit(false);PreparedStatement pstData = conn.prepareStatement("insert into stu_info values (?,?,?,?)");PreparedStatement pstOffset = conn.prepareStatement("insert into t_offsets values (?,?) on DUPLICATE KEY UPDATE offset = ?");// 需要把消费起始位置，初始化成上一次运行所记录的消费位移中去。// 而且还需要考虑一个问题，消费者再均衡时会发生什么（这是一个非常严重的问题！！！）启动的时候也是要发生再均衡的。// 这才是真正意义上的生产意义的代码consumer.subscribe(Arrays.asList("user-info"));boolean run = true;while(run){ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records)try{String data = record.value();// 解析原始数据String[] fields = data.split(",");// 插入mysql......// 执行业务数据插入语句pstData.execute();// 更新mysql中记录的偏移量pstOffset.setString(1,record.topic+":"+record.partition());pstOffset.setLong(2,record.offset()+1);pstOffset.setLong(3,record.offset()+1);pstOffset.execute();// 数据没提交，mysql自动回滚conn.commit();}catch(Exception e){conn.rollback();}}}}
}

起始位置的初始化、接续，要考虑两个环节：

1.程序启动时初始化

2.程序正常运行过程中发生了消费再均衡的过程，也需要进行起始位置的重新初始化

PreparedStatement pstQueryOffset = conn.prepareStatement("select offset from t_offsets where topic_partition = ?");consumer.subscribe(Arrays.asList("user-info"),new ConsumerRebalanceListener(){// 被剥夺分区消费权后会调用下面的方法public void onPartitionsRevoked(Collection<TopicPartition> partitions){// 如果某些场景下，不能用事务去收拾残局的话，可以在这个方法里面收拾}// 被分配了新的分区消费权后调用的方法public void onPartitionsAssigned(Collection<TopicPartition> partitions){try{// 去查询mysql 中的 t_offsets表，得到自己拥有消费权的分区的消费位移记录for(TopicPartition topicPartition : partitions){pstQueryOffset.setString(1,topicPartition.topic() + ":" + topicPartition.partition());ResultSet resultSet = pstQueryOffset.executeQuery();resultSet.next();long offset = resultSet.getLong("offset");// 将消费初始位置初始化为 数据库中查询到的偏移量consumer.seek（topicPartition,offset）；}}catch(Exception e){e.printStackTrace();}}
});

我们的这种方法比较彻底，根本就不用kafka去提交偏移量，而是将偏移量存储在mysql中。

consumer的消费位移提交方式：

1.全自动 auto.offset.commit = true; 定时提交到 consumer_offsets中。

2.半自动 auto.offset.commit = false; 然后手动触发提交，然后手动触发提交到 consumer.commitSync() -> 提交到consumer_offset中去。

提交到consumer_offset中的好处是初始化的时候会自动去找上一个offset

3.全手动 auto.offset.commit = false; 写自己的代码，去把消费者位移保存到你自己的地方 mysql等，将来初始化也需要自己自己去从自定义存储中查询到消费者的位移。

kafka 系统的 CAP 保证

分布式系统的 CAP 理论

CAP 理论作为分布式系统的基础理论,它描述的是一个分布式系统在以下三个特性中：

• 一致性（Consistency）
• 可用性（Availability）
• 分区容错性（Partition tolerance）

最多满足其中的两个特性。也就是下图所描述的。分布式系统要么满足 CA，要么 CP，要么 AP。无法同时满足 CAP。

分区容错性：

指的分布式系统中的某个节点或者网络分区出现了故障的时候，整个系统仍然能对外提供满足一致性和可用性的服务。也就是说部分故障不影响整体使用。

事实上我们在设计分布式系统是都会考虑到 bug,硬件，网络等各种原因造成的故障，所以即使部分节点或者网络出现故障，我们要求整个系统还是要继续使用的

(不继续使用,相当于只有一个分区,那么也就没有后续的一致性和可用性了)

可用性：

一直可以正常的做读写操作。简单而言就是客户端一直可以正常访问并得到系统的正常响应。用户角度来看就是不会出现系统操作失败或者访问超时等问题。

一致性：

在分布式系统完成某写操作后任何读操作，都应该获取到该写操作写入的那个最新的值。相当于要求分布式系统中的各节点时时刻刻保持数据的一致性。

---

kafka做的处理是，为了保证数据的一致性，数据的读写永远找分区里的leader，但是就会造成系统的可用性降低了，而kafka为了提升可用性，如果分区内其它副本挂了，其它的还可能成为leader，但是一致性可能降低，（因为有可能做不到完全同步）总之这三个特性很难完全的去满足。

Kafka 作为一个商业级消息中间件，数据可靠性和可用性是优先考虑的重点，兼顾数据一致性；

分区副本机制

kafka 从 0.8.0版本开始引入了分区版本，也就是说每个分区可以认为的配置几个副本（创建主题的时候指定replication-factor,也可以在broker级别进行配置 default.replication.factor）;

在众多的分区副本里面有一个副本是Leader，其余的副本是follower，所有的读写操作都是经过Leader进行的，同时follower会定期地去leader复制数据（通过心跳机制去请求同步数据）。当Leader挂了的时候，其中一个follower会重新成为新的Leader。通过分区服务，引入了数据冗余，同时也提供了kafka的数据可靠性。

Kafka的分区多副本架构是Kafka可靠性保证的核心，把消息写入多个副本可以使Kafka在发生崩溃时仍能保证消息的持久性。

分区副本的数据一致性困难

kafka 让分区多副本同步的基本手段是： follower 副本定期向 leader 请求数据同步！
既然是定期同步，则 leader 和 follower 之间必然存在各种数据不一致的情景！

问题 1：分区副本间动态不一致

问题 2：消费者所见不一致

如果此时 leader 宕机，follower1 或 follower2 被选为新的 leader，则 leader 换届前后，消费者所能读取到的数据发生了不一致；

问题 3：分区副本间最终不一致

一致性问题解决方案（HW）

动态过程中的副本数据不一致，是很难解决的；

kafka 先尝试着解决上述“消费者所见不一致”及“副本间数据最终不一致”的问题；

解决方案的核心思想

在动态不一致的过程中，维护一条步进式的“临时一致线”(既所谓的 High Watermark)

高水位线 HW = ISR 副本中最小 LEO（最大结束偏移量 + 1）

底层逻辑就是：offset < HW 的message，是各副本间一致的且安全的！

（如上图所示：offset < hw : 3 的message，是所有副本都已经备份好的数据)

高水位涉及到了数据的一致性。

leader中会记录 remoteLEO

等到在请求的时候，携带的LEO就会变了

此时leader这边的hw将会变成 6 ，再来请求的时候发给的就是6了

等到follow再次同步的时候，才能知道hw是多少

解决“消费者所见不一致” （消费者只允许看到 HW 以下的 message）

解决“分区副本数据最终不一致” （follower 数据按HW截断）

high watermark

代表数据在多副本备份的进度

hw就代表offset < hw的数据已经在所有isr副本间全部备份完毕。

所以，offset < hw的数据，让消费者可见，是相对安全的

HW 方案的天生缺陷

如前所述，看似 HW 解决了“分区数据最终不一致”的问题，以及“消费者所见不一致”的问题，但其实，这里面存在一个巨大的隐患，导致：

“分区数据最终不一致”的问题依然存在

producer 设置 acks=all 后，依然有可能丢失数据的问题

产生如上结果的根源是：HW 高水位线的更新，与数据同步的进度，存在迟滞！

Step 1：leader 和 follower 副本处于初始化值，follower 副本发送 fetch 请求，由于 leader 副本没有数据，因此不会进行同步操作；

Step 2：生产者发送了消息 m1 到分区 leader 副本，写入该条消息后 leader 更新 LEO = 1；

Step 3：follower 发送 fetch 请求，携带当前最新的 offset = 0，leader 处理 fetch 请求时，更新 remote LEO = 0，对比 LEO 值最小为 0，所以 HW = 0，leader 副本响应消息数据及 leader HW = 0 给follower，follower 写入消息后，更新 LEO 值，同时对比 leader HW 值，取最小的作为新的 HW 值，此时 follower HW = 0，这也意味着，follower HW 是不会超过 leader HW 值的。

Step 4：follower 发送第二轮 fetch 请求，携带当前最新的 offset = 1，leader 处理 fetch 请求时，更新 remote LEO = 1，对比 LEO 值最小为 1，所以 HW = 1，此时 leader 没有新的消息数据，所以直接返回 leader HW = 1 给 follower，follower 对比当前最新的 LEO 值 与 leader HW 值，取最小的作为新的 HW 值，此时 follower HW = 1。

从以上步骤可看出，leader 中保存的 remote LEO 值的更新（也即 HW 的更新）总是需要额外一轮fetch RPC 请求才能完成，这意味着在 leader 切换过程中，会存在数据丢失以及数据不一致的问题！

HW 会产生数据丢失和副本最终不一致问题

数据丢失的问题（即使 produce 设置 acks=all，依然会发生）

注意回顾：leader 中的 HW 值是在 follower 下一轮 fetch RPC 请求中完成更新的

如上图所示：

• 状态起始： B 为 leader，A 为 follower； 最新消息 m2 已同步，但 B 的 HW 比 A 的HW 大
• A 在此时崩溃（即 follower 没能通过下一轮请求来更新 HW 值）
• A 重启时，会自动将 LEO 值调整到之前的 HW 值，即会进行日志截断
• B 重启后，会从 向 A 发送 fetch 请求，收到 fetch 响应后，拿到 HW 值，并更新本地 HW 值，这时 B 会做日志截断，因此，offsets = 1 的消息被永久地删除了。

副本间数据最终不一致的问题（即使 produce 设置 acks=all，依然会发生）

如上图所示：

• 状态起始： A 为 leader，B 为 follower； 最新消息 m2 已同步，但 B 的 HW 比 A 的 HW 小
• A 在此时崩溃（即 follower 没能通过下一轮请求来更新 HW 值）
• B 先重启，会自动将 LEO 值调整到之前的 HW 值，即会进行日志截断，并在此刻接收了新的消息 m3，HW 随之上升为 2
• 然后，A 重启上线，会从 向 B 发送 fetch 请求，收到 fetch 响应后，拿到 HW 值，并更新本地 HW 值，发现不需要截断，从而已经产生了“副本间数据最终不一致”！

只要新一届 leader 在老 leader 重启上线前，接收了新的数据，就可能发生上图中的场景，根源也在于HW 的更新落后于数据同步进度

Leader-Epoch 机制的引入

为了解决 HW 更新时机是异步延迟的，而 HW 又是决定日志是否备份成功的标志，从而造成数据丢失和数据不一致的现象，Kafka 引入了 leader epoch 机制；

在每个副本日志目录下都创建一个 leader-epoch-checkpoint 文件，用于保存 leader 的 epoch 信息

leader-epoch 的含义

如下，leader epoch 长这样：

它的格式为 (epoch offset)，epoch 指的是 leader 版本，它是一个单调递增的一个正整数值，每次 leader变更，epoch 版本都会 +1，offset 是每一代 leader 写入的第一条消息的位移值，比如：

(0,0)
(1,300)

以上第 2 个版本是从位移 300 开始写入消息，意味着第一个版本写入了 0-299 的消息。

这里面记录信息的本质试：从哪个offset开始的数据，是从那届leader写入的。

leader epoch 具体的工作机制：

• 当副本成为 leader 时：

这时，如果此时生产者有新消息发送过来，会首先更新 leader epoch 以及 LEO ，并添加到leader-epoch-checkpoint 文件中

• 当副本变成 follower 时

发送 LeaderEpochRequest 请求给 leader 副本，该请求包括了 follower 中最新的 epoch 版本；leader 返回给 follower 的响应中包含了一个 LastOffset，如果 follower last epoch = leader last epoch（纪元相同），则 LastOffset = leader LEO，否则取 大于 follower last epoch 中最小的 leader epoch 的 start offset值；

举个例子：假设 follower last epoch = 1，此时 leader 有 (1, 20) (2, 80) (3, 120)，则 LastOffset = 80；

follower 拿到 LastOffset 之后，会对比当前 LEO 值是否大于 LastOffset，如果当前 LEO 大于LastOffset，则从 LastOffset 截断日志；follower 开始发送 fetch 请求给 leader 保持消息同步。

leader epoch 如何解决 HW 的备份缺陷

解决数据丢失：

如上图所示：

A 重启之后，发送 LeaderEpochRequest 请求给 B，由于 B 还没追加消息，此时 epoch = request epoch = 0，因此返回 LastOffset = leader LEO = 2 给 A

A 拿到 LastOffset 之后，发现等于当前 LEO 值，故不用进行日志截断。就在这时 B 宕机了，A 成为 leader，在 B 启动回来后，会重复 A 的动作，同样不需要进行日志截断，数据没有丢失。

解决数据最终不一致问题：

如上图所示:

• A 和 B 同时宕机后，B 先重启回来成为分区 leader，这时候生产者发送了一条消息过来，leader epoch 更新到 1
• 此时 A 启动回来后，发送 LeaderEpochReques（t follower epoch = 0）给 B，B 判断 follower epoch不等于最新的 epoch，于是找到大于 follower epoch 最小的 epoch = 1，即 LastOffset = epoch start offset = 1
• A 拿到 LastOffset 后，判断小于当前 LEO 值，于是从 LastOffset 位置进行日志截断，接着开始发送 fetch 请求给 B 开始同步消息，避免了消息不一致/离散的问题。

---

HW的时候，计算producer把acks = all，依然会丢数据，因为它是依赖HW来进行数据截断的。

而HW的更新是相对于数据同步进度落后一轮请求的。

而现在，acks = all依赖于 leader epoch的话，不会再有数据丢失发生了，也不会再有消息不一致情况了。

LEO/HW/LSO 等相关术语速查

LEO:（last end offset）就是该副本中消息的最大偏移量的值+1

HW:（high watermark）各副本中 LEO 的最小值。这个值规定了消费者仅能消费 HW 之前的数据

LEO 与 HW 与数据一致性密切相关；

如图，各副本中最小的 LEO 是 3，所以 HW 是 3，所以，消费者此刻最多能读Msg2;

不清洁选举[了解]

不清洁选举，是指允许“非 ISR 副本”可以被选举为 leader；非 ISR 副本被选举为 leader，将极大增加数据丢失及数据不一致的可能性！由参数 unclean.leader.election.enable=false（默认） 控制；

初始状态： follower2 严重落后于 leader，并且不属于 ISR 副本

此刻，所有 ISR 副本宕机

Follower2 成为新的 leader，并接收数据

之前宕机副本重启，按照最新 leader 的最新 leo 进行截断，产生数据丢失及不一致