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

2023-12-15 07:39:50


本章主题是: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 写入的所有消息。

文章来源:https://blog.csdn.net/fujuacm/article/details/134974036
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。