8、Kafka 消息交付可靠性保障（幂等生产者和事务生产者是一回事吗？）

本章主题是：Kafka 消息交付可靠性保障以及精确处理一次语义的实现。
实用场景如 支付成功后的短信通知（重复提示，会让用户以为支付了多次）。

所谓的消息交付可靠性保障，是指 Kafka 对 Producer 和 Consumer 要处理的消息提供什么样的承诺。常见的承诺有以下三种：

• 至少一次（at least once）：消息不会丢失，但有可能被重复发送。
• 最多一次（at most once）：消息可能会丢失，但绝不会被重复发送。
• 精确一次（exactly once）：消息不会丢失，也不会被重复发送。

1、至少一次（at least once）交付可靠性保障

至少一次（at least once）：消息不会丢失，但有可能被重复发送。
这是 Kafka 默认提供的交付可靠性保障。

在这个保障下，Kafka 认为某条消息发送成功有 2 点要求

• 1、Broker 成功 “提交” 消息
• 2、Producer 接到 Broker “提交”后的应答

那在这 2 个要求下，为什么会导致消息被重复发送呢？
因为倘若消息成功 “提交”，但 Broker 的应答没有成功发送回 Producer 端（比如网络出现瞬时抖动），那么 Producer 会重试发送，因为它无法确定消息是否真的提交成功了。然后就会导致消息重复发送，但也保证了消息不丢失。

2、最多一次（at most once）交付可靠性保障

最多一次（at most once）：消息可能会丢失，但绝不会被重复发送。
在上面讲到，至少一次会导致消息重复发送，是 Producer 会重试发送，因为其未接收到 Broker 的“提交”应答。
那相应地，只需要让 Producer 禁止重试即可。这样一来，消息要么写入成功，要么写入失败。就不会重复发送，但可能会导致消息丢失。

3、精确一次（exactly once）

精确一次（exactly once）：消息不会丢失，也不会被重复发送。

Kafka 是怎么做到精确一次的呢？自 kafka 0.11 版本开始，实现方式是通过两种机制：幂等性（Idempotence）和事务（Transaction）。

3.1、幂等性

3.1.1、什么是幂等性（Idempotence）？

幂等性指若一个子程序是幂等的，那它必然不能修改系统状态。这样不管运行这个子程序多少次，与该子程序关联的那部分系统状态保持不变。

幂等性有很多好处，其最大的优势在于我们可以安全地重试任何幂等性操作，反正它们也不会破坏我们的系统状态。如果是非幂等性操作，我们还需要担心某些操作执行多次对状态的影响，但对于幂等性操作而言，我们根本无需担心此事。

3.1.2、幂等型 Producer

Producer 默认不是幂等性的，但我们可以创建幂等型 Producer。这是 0.11.0.0 版本引入的新功能。

仅需要设置一个参数即可实现：enable.idempotence=true。此时 Kafka 自动帮你做消息的重复去重。主要是给每条消息附上分布式唯一id，这是常规的分布式幂等方案。

分布式唯一Id具体规则
Kafka 每条消息的唯一ID由两部分组成：每个生产者自己分配到的唯一ID拼接每条消息自带的序列号。从0.11版本开始，Kafka为每个Producer端分配了一个唯一的ID，同时每条消息中也会携带一个序列号，这样服务端就可以对消息进行去重。

3.1.3、幂等机制的作用范围

幂等机制的作用范围有如下 2 个限制

• 它只能保证单分区上的幂等性，即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息，它无法实现多个分区的幂等性。
• 它只能实现单会话上的幂等性，不能实现跨会话的幂等性。这里的会话，你可以理解为 Producer 进程的一次运行。当你重启了 Producer 进程之后，这种幂等性保证就丧失了。

那么你可能会问，如果我想实现多分区以及多会话上的消息无重复，应该怎么做呢？答案就是事务（transaction）或者依赖事务型 Producer。这也是幂等性 Producer 和事务型 Producer 的最大区别！

3.2、事务

3.2.1、什么是事务

kafka 自 0.11 版本开始也提供了对事务的支持，目前主要是在 read committed 隔离级别上做事情。它能保证多条消息原子性地写入到目标分区，同时也能保证 Consumer 只能看到事务成功提交的消息。

3.2.2、事务型 Producer

作用：事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功，要么全部失败。另外，事务型 Producer 也不惧进程的重启。Producer 重启回来后，Kafka 依然保证它们发送消息的精确一次处理。

实现：
设置事务型 Producer 的方法只要做 2 点即可：

• 和幂等性 Producer 一样，开启 enable.idempotence = true。
• 设置 Producer 端参数 transactional.id。最好为其设置一个有意义的名字。

然后在 Producer 代码显著调用一些事务 API，如 initTransaction、beginTransaction、commitTransaction 和 abortTransaction，它们分别对应事务的初始化、事务开始、事务提交以及事务终止。

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

这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka，要么它们全部提交成功，要么全部写入失败。实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更的。修改起来也很简单，设置 isolation.level 参数的值即可。当前这个参数有两个取值：

• read_uncommitted：这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。很显然，如果你用了事务型 Producer，那么对应的 Consumer 就不要使用这个值。
• read_committed：表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了，它也能看到非事务型 Producer 写入的所有消息。