下面我们来简单了解一下消息传递语义,以及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。
图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。
图 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只能保证单分区上无重复消息;事务可以保证多分区写入消息的完整性。
图 事务
这样producer端实现了exactly once,那么consumer端呢?
consumer端由于可能无法消费事务中所有消息,并且消息可能被删除,所以事务并不能解决consumer端exactly once的问题,我们可能还是需要自己处理这方面的逻辑。比如自己管理offset的提交,不要自动提交,也是可以实现exactly once的。
还有一个选择就是使用kafka自己的流处理引擎,也就是Kafka Streams,
设置processing.guarantee=exactly_once,就可以轻松实现exactly once了。