Kafka消息送达语义详解
消息送达语义是消息系统中一个常见的问题，主要包含三种语义：
- At most once：消息发送或消费至多一次
- At least once：消息发送或消费至少一次
- Exactly once：消息恰好只发送一次或只消费一次
下面我们分别从发送者和消费者的角度来阐述这三种消息送达语义。
Producer
从Producer的角度来看，At most once意味着Producer发送完一条消息后，不会确认消息是否成功送达。这样从Producer的角度来看，消息仅仅被发送一次，也就存在者丢失的可能性。
从Producer的角度来看，At least once意味着Producer发送完一条消息后，会确认消息是否发送成功。如果Producer没有收到Broker的ack确认消息，那么会不断重试发送消息。这样就意味着消息可能被发送不止一次，也就存在这消息重复的可能性。
从Producer的角度来看，Exactly once意味着Producer消息的发送是幂等的。这意味着不论消息重发多少遍，最终Broker上记录的只有一条不重复的数据。
Producer At least once配置
Kafka默认的Producer消息送达语义就是At least once，这意味着我们不用做任何配置就能够实现At least once消息语义。
原因是Kafka中默认
acks=1并且retries=2147483647。
Producer At most once配置
我们可以通过配置Producer的以下配置项来实现At most once语义：
-
acks=0。
acks配置项表示Producer期望的Broker的确认数。默认值为1。可选项：[0，1，all]。如果设置为0，表示Producer发送完消息后不会等待任何Broker的确认；设置为1表示Producer会等待Broker集群中的leader的确认写入消息；设置为all表示Producer需要等待Broker集群中leader和其所有follower的确认写入消息。 -
retries=0。
retires配置项表示当消息发送失败时，Producer重发消息的次数。默认值为2147483647。当配置了acks=0时，retries配置项就失去了作用，因此这儿可以不用配置。
当配置了
retires的值后，如果没有将max.in.flight.requests.per.connection配置的值设置为1，有可能造成消息乱序的结果。max.in.flight.requests.per.connection配置代表着一个Producer同时可以发送的未收到确认的消息数量。如果max.in.flight.requests.per.connection数量大于1，那么可能发送了message1后，在没有收到确认前就发送了message2，此时message1发送失败后触发重试，而message2直接发送成功，就造成了Broker上消息的乱序。max.in.flight.requests.per.connection的默认值为5。
Producer Exactly once配置
Exactly once是Kafka从版本0.11之后提供的高级特性。我们可以通过配置Producer的以下配置项来实现Exactly once语义：
-
enable.idempotence=true。
enable.idempotence配置项表示是否使用幂等性。当enable.idempotence配置为true时，acks必须配置为all。并且建议max.in.flight.requests.per.connection的值小于5。 - acks=all。
Kafka如何实现消息发送幂等
Kafka本身支持At least once消息送达语义，因此实现消息发送的幂等关键是要实现Broker端消息的去重。为了实现消息发送的幂等性，Kafka引入了两个新的概念：
- PID。每个新的Producer在初始化的时候会被分配一个唯一的PID，这个PID对用户是不可见的。
- Sequence Numbler。对于每个PID，该Producer发送数据的每个<Topic, Partition>都对应一个从0开始单调递增的Sequence Number
Broker端在缓存中保存了这Sequence Numbler，对于接收的每条消息，如果其序号比Broker缓存中序号大于1则接受它，否则将其丢弃。这样就可以实现了消息重复提交了。但是，只能保证单个Producer对于同一个<Topic, Partition>的Exactly Once语义。不能保证同一个Producer一个topic不同的partion幂等。
Kafka幂等性配置时要求
max.in.flight.requests.per.connection小于等于 5 的主要原因是：Server 端的 ProducerStateManager 实例会缓存每个 PID 在每个 Topic-Partition 上发送的最近 5 个batch 数据（这个 5 是写死的，至于为什么是 5，可能跟经验有关，当不设置幂等性时，当这个设置为 5 时，性能相对来说较高，社区是有一个相关测试文档，忘记在哪了），如果超过 5，ProducerStateManager 就会将最旧的 batch 数据清除。假设应用将 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION 设置为 6，假设发送的请求顺序是 1、2、3、4、5、6，这时候 server 端只能缓存 2、3、4、5、6 请求对应的 batch 数据，这时候假设请求 1 发送失败，需要重试，当重试的请求发送过来后，首先先检查是否为重复的 batch，这时候检查的结果是否，之后会开始 check 其 sequence number 值，这时候只会返回一个 OutOfOrderSequenceException 异常，client 在收到这个异常后，会再次进行重试，直到超过最大重试次数或者超时，这样不但会影响 Producer 性能，还可能给 Server 带来压力（相当于client 狂发错误请求）。
Consumer
从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也能回滚。
Consumer At least once配置
- enable.auto.commit=false。禁止后台自动提交offset。
- 手动调用
consumer.commitSync()来提交offset。手动调用保证了offset即时更新。
通过手动提交offset，就可以实现Consumer At least once语义。
Consumer At most once配置
- enable.auto.commit=true。后台定时提交offset。
-
auto.commit.interval.ms配置为一个很小的数值。
auto.commit.interval.ms表示后台提交offset的时间间隔。
通过自动提交offset，并且将定时提交时间间隔设置的很小，就可以实现Consumer At most once语义。
Consumer Exactly once配置
-
isolation.level=read_committed。
isolation.level表示何种类型的message对Consumer可见。下面会更详细的介绍这个参数。
一个常见的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的事务语义。
