Question

Kafka消息送达语义详解

Answer 1

消息送达语义 是消息系统中一个常见的问题，主要包含三种语义：

下面我们分别从发送者和消费者的角度来阐述这三种消息送达语义。

从Producer的角度来看， At most once意味着Producer发送完一条消息后，不会确认消息是否成功送达 。这样从Producer的角度来看，消息仅仅被发送一次，也就存在者丢失的可能性。

从Producer的角度来看， At least once意味着Producer发送完一条消息后，会确认消息是否发送成功。如果Producer没有收到Broker的ack确认消息，那么会不断重试发送消息 。这样就意味着消息可能被发送不止一次，也就存在这消息重复的可能性。

从Producer的角度来看， Exactly once意味着Producer消息的发送是幂等的 。这意味着不论消息重发多少遍，最终Broker上记录的只有一条不重复的数据。

Kafka默认的Producer消息送达语义就是At least once，这意味着我们不用做任何配置就能够实现At least once消息语义。

我们可以通过配置Producer的以下配置项来实现At most once语义：

Exactly once是Kafka从版本0.11之后提供的高级特性。我们可以通过配置Producer的以下配置项来实现Exactly once语义：

Kafka本身支持At least once消息送达语义，因此实现消息发送的幂等关键是要实现Broker端消息的去重。为了实现消息发送的幂等性，Kafka引入了两个新的概念：

Broker端在缓存中保存了这Sequence Numbler，对于接收的每条消息，如果其序号比Broker缓存中序号大于1则接受它，否则将其丢弃。这样就可以实现了消息重复提交了。但是，只能保证单个Producer对于同一个<Topic, Partition>的Exactly Once语义。不能保证同一个Producer一个topic不同的partion幂等。

从Consumer的角度来看， At most once意味着Consumer对一条消息最多消费一次，因此有可能存在消息消费失败依旧提交offset的情况 。考虑下面的情况：Consumer首先读取消息，然后提交offset，最后处理这条消息。在处理消息时，Consumer宕机了，此时offset已经提交，下一次读取消息时读到的是下一条消息了，这就是At most once消费。

从Consumer的角度来看， At least once意味着Consumer对一条消息可能消费多次 。考虑下面的情况：Consumer首先读取消息，然后处理这条消息，最后提交offset。在处理消息时成功后，Consumer宕机了，此时offset还未提交，下一次读取消息时依旧是这条消息，那么处理消息的逻辑又将被执行一遍，这就是At least once消费。

从Consumer的角度来看， Exactly once意味着消息的消费处理逻辑和offset的提交是原子性的，即消息消费成功后offset改变，消息消费失败offset也能回滚 。

通过手动提交offset，就可以实现Consumer At least once语义。

通过自动提交offset，并且将定时提交时间间隔设置的很小，就可以实现Consumer At most once语义。

一个常见的Exactly once的的使用场景是：当我们订阅了一个topic，然后往另一个topic里写入数据时，我们希望这两个操作是原子性的，即如果写入消息失败，那么我们希望读取消息的offset可以回滚。

此时可以通过Kafka的Transaction特性来实现。Kafka是在版本0.11之后开始提供事务特性的。我们可以将Consumer读取数据和Producer写入数据放进一个同一个事务中，在事务没有成功结束前，所有的这个事务中包含的消息都被标记为uncommitted。只有事务执行成功后，所有的消息才会被标记为committed。

我们知道，offset信息是以消息的方式存储在Broker的__consumer_offsets topic中的。因此在事务开始后，Consumer读取消息后，所有的offset消息都是uncommitted状态。所有的Producer写入的消息也都是uncommitted状态。

而Consumer可以通过配置 isolation.level 来决定uncommitted状态的message是否对Consumer可见。 isolation.level 拥有两个可选值： read_committed 和 read_uncommitted 。默认值为 read_uncommitted 。

当我们将 isolation.level 配置为 read_committed 后，那么所有事务未提交的数据就都对Consumer不可见了，也就实现了Kafka的事务语义。