Kafka的消息会丢失和重复吗?——如何实现Kafka精确传递一次语义

简介: 我们都知道Kafka的吞吐量很大,但是Kafka究竟会不会丢失消息呢?又会不会重复消费消息呢?有很多公司因为业务要求必须保证消息不丢失、不重复的到达,比如无人机实时监控系统,当无人机闯入机场区域,我们必须立刻报警,不允许消息丢失。而无人机离开禁飞区域后我们需要将及时报警解除。如果消息重复了呢,我们是否需要复杂的逻辑来自己处理消息重复的情况呢,这种情况恐怕相当复杂而难以处理。但是如果我们能保证消息exactly once,那么一切都容易得多。

下面我们来简单了解一下消息传递语义,以及kafka的消息传递机制。

首先我们要了解的是message delivery semantic   也就是消息传递语义。

这是一个通用的概念,也就是消息传递过程中消息传递的保证性。

分为三种:

最多一次(at most once): 消息可能丢失也可能被处理,但最多只会被处理一次。

可能丢失 不会重复

至少一次(at least once):  消息不会丢失,但可能被处理多次。

可能重复 不会丢失

精确传递一次(exactly once): 消息被处理且只会被处理一次。

不丢失 不重复 就一次

而kafka其实有两次消息传递,一次生产者发送消息给kafka,一次消费者去kafka消费消息。

两次传递都会影响最终结果,

两次都是精确一次,最终结果才是精确一次。

两次中有一次会丢失消息,或者有一次会重复,那么最终的结果就是可能丢失或者重复的。


一、Produce端消息传递


这是producer端的代码:

Properties properties = new Properties();
        properties.put("bootstrap.servers", "kafka01:9092,kafka02:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for (int i = 1; i <= 600; i++) {
            kafkaProducer.send(new ProducerRecord<String, String>("z_test_20190430", "testkafka0613"+i));
            System.out.println("testkafka"+i);
        }
        kafkaProducer.close();

其中指定了一个参数acks   可以有三个值选择:

0:producer完全不管broker的处理结果 回调也就没有用了 并不能保证消息成功发送 但是这种吞吐量最高

all或者-1:leader broker会等消息写入 并且ISR都写入后 才会响应,这种只要ISR有副本存活就肯定不会丢失,但吞吐量最低。

1:默认的值 leader broker自己写入后就响应,不会等待ISR其他的副本写入,只要leader broker存活就不会丢失,即保证了不丢失,也保证了吞吐量。

所以设置为0时,实现了at most once,而且从这边看只要保证集群稳定的情况下,不设置为0,消息不会丢失。

但是还有一种情况就是消息成功写入,而这个时候由于网络问题producer没有收到写入成功的响应,producer就会开启重试的操作,直到网络恢复,消息就发送了多次。这就是at least once了。

kafka producer 的参数acks 的默认值为1,所以默认的producer级别是at least once。并不能exactly once。

微信图片_20220525224431.png

图kafka-apis


二、Consumer端消息传递


consumer是靠offset保证消息传递的。

consumer消费的代码如下:

Properties props = new Properties();
        props.put("bootstrap.servers", "kafka01:9092,kafka02:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset","earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("foo", "bar"));
      try{
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
         }
        }finally{
          consumer.close();
        }

其中有一个参数是  enable.auto.commit

若设置为true   consumer在消费之前提交位移 就实现了at most once

若是消费后提交 就实现了 at least once  默认的配置就是这个。

kafka consumer的参数enable.auto.commit的默认值为true   ,所以默认的consumer级别是at least once。也并不能exactly once。

微信图片_20220525224435.png

图 consumer-groups


三、精确一次


通过了解producer端与consumer端的设置,我们发现kafka在两端的默认配置都是at least once,肯能重复,通过配置的话呢也不能做到exactly once,好像kafka的消息一定会丢失或者重复的,是不是没有办法做到exactly once了呢?

确实在kafka 0.11.0.0版本之前producer端确实是不可能的,但是在kafka 0.11.0.0版本之后,kafka正式推出了idempotent  producer。

也就是幂等的producer还有对事务的支持。

幂等的producer

kafka 0.11.0.0版本引入了idempotent  producer机制,在这个机制中同一消息可能被producer发送多次,但是在broker端只会写入一次,他为每一条消息编号去重,而且对kafka开销影响不大。

如何设置开启呢? 需要设置producer端的新参数  enable.idempotent  为true。

而多分区的情况,我们需要保证原子性的写入多个分区,即写入到多个分区的消息要么全部成功,要么全部回滚。

这时候就需要使用事务,在producer端设置 transcational.id为一个指定字符串。

这样幂等producer只能保证单分区上无重复消息;事务可以保证多分区写入消息的完整性。

微信图片_20220525224438.jpg

图 事务

这样producer端实现了exactly once,那么consumer端呢?

consumer端由于可能无法消费事务中所有消息,并且消息可能被删除,所以事务并不能解决consumer端exactly once的问题,我们可能还是需要自己处理这方面的逻辑。比如自己管理offset的提交,不要自动提交,也是可以实现exactly once的。

还有一个选择就是使用kafka自己的流处理引擎,也就是Kafka Streams,

设置processing.guarantee=exactly_once,就可以轻松实现exactly once了。

相关文章
|
7月前
|
消息中间件 Kafka API
Kafka Exactly Once 语义实现原理:幂等性与事务消息
Apache Kafka的Exactly-Once语义确保了消息处理的准确性和一致性。通过幂等性和事务消息,Kafka实现了要么全处理要么全不处理的原子性。文章详细解析了Kafka事务的工作流程,包括生产者的幂等性(通过序列号保证),以及事务消息的提交和回滚过程。Kafka事务提供了ACID保证,但存在性能限制,如额外的RPC请求和单生产者只能执行一个事务。此外,事务适用于同集群内的操作,跨集群时原子性无法保证。了解这些原理有助于开发者更好地利用Kafka事务构建可靠的数据处理系统。
185 3
 Kafka Exactly Once 语义实现原理:幂等性与事务消息
|
消息中间件 Kafka
Kafka对于消息顺序性的最佳实践
Kafka对于消息顺序性的最佳实践
|
消息中间件 存储 缓存
不看损失大了,刨根问底,Kafka消息中间件到底会不会丢消息
不看损失大了,刨根问底,Kafka消息中间件到底会不会丢消息
不看损失大了,刨根问底,Kafka消息中间件到底会不会丢消息
|
消息中间件 存储 缓存
Kafka快速入门(生产者)同步异步发送、分区、消息精确一次发送、幂等性、事务
Kafka快速入门(生产者)同步异步发送、分区、消息精确一次发送、幂等性、事务
Kafka快速入门(生产者)同步异步发送、分区、消息精确一次发送、幂等性、事务
|
消息中间件 监控 Kafka
发送kafka消息的shell脚本
开发和学习时需要造一些kafka消息,于是写了段脚本实现,在这里记录备忘,后面会常用到
396 0
发送kafka消息的shell脚本
|
消息中间件 存储 SQL
阿里云消息队列 Kafka-消息检索实践
本文章主要介绍消息队列使用过程中所遇到的消息丢失、重复消费等痛点问题的排查办法,以及消息队列 Kafka「检索组件」的场景实践,并对其关键技术进行解读。旨在帮助大家对消息队列 Kafka「检索组件」的特点和使用方式更加熟悉,以更有效地解决消息排查过程中所遇到的问题。
阿里云消息队列 Kafka-消息检索实践
|
消息中间件 存储 Kafka
kafka消息丢失的场景分析
kafka消息丢失的场景分析
364 0
kafka消息丢失的场景分析
|
消息中间件 运维 Kafka
【kafka问题】记一次kafka消费者未接收到消息问题
出现了这样一个问题, A说他的kafka消息发送了; B说它没有接收到; 那么问题来了: • A的消息是否发送了? • 如果A的消息发送成功了; B为何没有消费到? 好,带着上面的问题,我们来一步步排查一下问题所在
【kafka问题】记一次kafka消费者未接收到消息问题
|
消息中间件 存储 负载均衡
Zookeeper搭载kafka消息发布和订阅
ZooKeeper 是一个开源的分布式协调服务,由雅虎创建,是 Google Chubby 的开源实现。 分布式应用程序可以基于 ZooKeeper 实现诸如数据发布/订阅、负载均衡、命名服务、分布式协调/通知、 集群管理、Master 选举、配置维护,名字服务、分布式同步、分布式锁和分布式队列等功能。
181 0
|
消息中间件 存储 缓存
双十一期间Kafka以这种方式丢消息让我猝不及防
双十一期间Kafka以这种方式丢消息让我猝不及防
双十一期间Kafka以这种方式丢消息让我猝不及防

热门文章

最新文章

下一篇
无影云桌面