kafka实现无消息丢失与精确一次语义(exactly once)处理 (2)

消费者丢失的情况,其实跟消费者位移处理不当有关。消费者位移提交有一个参数,enable.auto.commit,默认是true,决定是否要让消费者自动提交位移。如果开启,那么consumer每次都是先提交位移,再进行消费,比如先跟broker说这5个数据我消费好了,然后才开始慢慢消费这5个数据。

这样处理的话,好处是简单,坏处就是漏消费数据,比如你说要消费5个数据,消费了2个自己就挂了。那下次该consumer重启后,在broker的记录中这个consumer是已经消费了5个的。

所以最好的做法就是将enable.auto.commit设置为false,改为手动提交位移,在每次消费完之后再手动提交位移信息。当然这样又有可能会重复消费数据,毕竟exactly once处理一直是一个问题呀(/摊手)。遗憾的是kafka目前没有保证consumer幂等消费的措施,如果确实需要保证consumer的幂等,可以对每条消息维持一个全局的id,每次消费进行去重,当然耗费这么多的资源来实现exactly once的消费到底值不值,那就得看具体业务了。

1.4 无消息丢失小结

那么到这里先来总结下无消息丢失的主要配置吧:

producer的acks设置位-1,同时min.insync.replicas设置大于1。并且使用带有回调的producer api发生消息。

默认副本数replication.factor设置为大于1,或者创建topic的时候指定大于1的副本数。

unclean.leader.election.enable 设置为false,防止定期副本leader重选举

消费者端,自动提交位移enable.auto.commit设置为false。在消费完后手动提交位移。

那么接下来就来说说kafka实现精确一次(exactly once)处理的方法吧。

实现精确一次(exactly once)处理

在分布式环境下,要实现消息一致与精确一次(exactly once)语义处理是很难的。精确一次处理意味着一个消息只处理一次,造成一次的效果,不能多也不能少。

那么kafka如何能够实现这样的效果呢?在介绍之前,我们先来介绍其他两个语义,至多一次(at most once)和至少一次(at least once)。

最多一次和至少一次

最多一次就是保证一条消息只发送一次,这个其实最简单,异步发送一次然后不管就可以,缺点是容易丢数据,所以一般不采用。

至少一次语义是kafka默认提供的语义,它保证每条消息都能至少接收并处理一次,缺点是可能有重复数据。

前面有介绍过acks机制,当设置producer客户端的acks是1的时候,broker接收到消息就会跟producer确认。但producer发送一条消息后,可能因为网络原因消息超时未达,这时候producer客户端会选择重发,broker回应接收到消息,但很可能最开始发送的消息延迟到达,就会造成消息重复接收。

那么针对这些情况,要如何实现精确一次处理的语义呢?

幂等的producer

要介绍幂等的producer之前,得先了解一下幂等这个词是什么意思。幂等这个词最早起源于函数式编程,意思是一个函数无论执行多少次都会返回一样的结果。比如说让一个数加1就不是幂等的,而让一个数取整就是幂等的。因为这个特性所以幂等的函数适用于并发的场景下。

但幂等在分布式系统中含义又做了进一步的延申,比如在kafka中,幂等性意味着一个消息无论重复多少次,都会被当作一个消息来持久化处理。

kafka的producer默认是支持最少一次语义,也就是说不是幂等的,这样在一些比如支付等要求精确数据的场景会出现问题,在0.11.0后,kafka提供了让producer支持幂等的配置操作。即:

props.put("enable.idempotence", ture)

在创建producer客户端的时候,添加这一行配置,producer就变成幂等的了。注意开启幂等性的时候,acks就自动是“all”了,如果这时候手动将ackss设置为0,那么会报错。

而底层实现其实也很简单,就是对每条消息生成一个id值,broker会根据这个id值进行去重,从而实现幂等,这样一来就能够实现精确一次的语义了。

但是!幂等的producery也并非万能。有两个主要是缺陷:

幂等性的producer仅做到单分区上的幂等性,即单分区消息不重复,多分区无法保证幂等性。

只能保持单会话的幂等性,无法实现跨会话的幂等性,也就是说如果producer挂掉再重启,无法保证两个会话间的幂等(新会话可能会重发)。因为broker端无法获取之前的状态信息,所以无法实现跨会话的幂等。

事务的producer

当遇到上述幂等性的缺陷无法解决的时候,可以考虑使用事务了。事务可以支持多分区的数据完整性,原子性。并且支持跨会话的exactly once处理语义,也就是说如果producer宕机重启,依旧能保证数据只处理一次。

开启事务也很简单,首先需要开启幂等性,即设置enable.idempotence为true。然后对producer发送代码做一些小小的修改。

//初始化事务 producer.initTransactions(); try { //开启一个事务 producer.beginTransaction(); producer.send(record1); producer.send(record2); //提交 producer.commitTransaction(); } catch (KafkaException e) { //出现异常的时候,终止事务 producer.abortTransaction(); }

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/zysdys.html