消息送达语义是消息系统中一个常见的问题,主要包含三种语义:
- 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的事务语义。