如何實現(xiàn)Kafka精確傳遞一次語義

有很多公司因為業(yè)務(wù)要求必須保證消息不丟失、不重復(fù)的到達(dá)，比如無人機(jī)實時監(jiān)控系統(tǒng)，當(dāng)無人機(jī)闖入機(jī)場區(qū)域，我們必須立刻報警，不允許消息丟失。而無人機(jī)離開禁飛區(qū)域后我們需要將及時報警解除。如果消息重復(fù)了呢，我們是否需要復(fù)雜的邏輯來自己處理消息重復(fù)的情況呢，這種情況恐怕相當(dāng)復(fù)雜而難以處理。但是如果我們能保證消息exactly once，那么一切都容易得多。

下面我們來簡單了解一下消息傳遞語義，以及kafka的消息傳遞機(jī)制。

首先我們要了解的是message delivery semantic   也就是消息傳遞語義。

這是一個通用的概念，也就是消息傳遞過程中消息傳遞的保證性。

分為三種：

最多一次（at most once）: 消息可能丟失也可能被處理，但最多只會被處理一次。

可能丟失 不會重復(fù)

至少一次（at least once）:  消息不會丟失，但可能被處理多次。

可能重復(fù) 不會丟失

精確傳遞一次（exactly once）: 消息被處理且只會被處理一次。

不丟失 不重復(fù) 就一次

而kafka其實有兩次消息傳遞，一次生產(chǎn)者發(fā)送消息給kafka，一次消費者去kafka消費消息。

兩次傳遞都會影響最終結(jié)果，

兩次都是精確一次，最終結(jié)果才是精確一次。

兩次中有一次會丟失消息，或者有一次會重復(fù)，那么最終的結(jié)果就是可能丟失或者重復(fù)的。

一、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();

其中指定了一個參數(shù)acks   可以有三個值選擇：

0：producer完全不管broker的處理結(jié)果 回調(diào)也就沒有用了 并不能保證消息成功發(fā)送 但是這種吞吐量最高

all或者-1：leader broker會等消息寫入 并且ISR都寫入后 才會響應(yīng)，這種只要ISR有副本存活就肯定不會丟失，但吞吐量最低。

1：默認(rèn)的值 leader broker自己寫入后就響應(yīng)，不會等待ISR其他的副本寫入，只要leader broker存活就不會丟失，即保證了不丟失，也保證了吞吐量。

所以設(shè)置為0時，實現(xiàn)了at most once，而且從這邊看只要保證集群穩(wěn)定的情況下，不設(shè)置為0，消息不會丟失。

但是還有一種情況就是消息成功寫入，而這個時候由于網(wǎng)絡(luò)問題producer沒有收到寫入成功的響應(yīng)，producer就會開啟重試的操作，直到網(wǎng)絡(luò)恢復(fù)，消息就發(fā)送了多次。這就是at least once了。

kafka producer 的參數(shù)acks 的默認(rèn)值為1，所以默認(rèn)的producer級別是at least once。并不能exactly once。

二、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();
        }

其中有一個參數(shù)是  enable.auto.commit

若設(shè)置為true   consumer在消費之前提交位移 就實現(xiàn)了at most once

若是消費后提交 就實現(xiàn)了 at least once  默認(rèn)的配置就是這個。

kafka consumer的參數(shù)enable.auto.commit的默認(rèn)值為true   ，所以默認(rèn)的consumer級別是at least once。也并不能exactly once。

三、精確一次

通過了解producer端與consumer端的設(shè)置，我們發(fā)現(xiàn)kafka在兩端的默認(rèn)配置都是at least once，肯能重復(fù)，通過配置的話呢也不能做到exactly once，好像kafka的消息一定會丟失或者重復(fù)的，是不是沒有辦法做到exactly once了呢？

確實在kafka 0.11.0.0版本之前producer端確實是不可能的，但是在kafka 0.11.0.0版本之后，kafka正式推出了idempotent  producer。

也就是冪等的producer還有對事務(wù)的支持。

冪等的producer

kafka 0.11.0.0版本引入了idempotent  producer機(jī)制，在這個機(jī)制中同一消息可能被producer發(fā)送多次，但是在broker端只會寫入一次，他為每一條消息編號去重，而且對kafka開銷影響不大。

如何設(shè)置開啟呢？ 需要設(shè)置producer端的新參數(shù)  enable.idempotent  為true。

而多分區(qū)的情況，我們需要保證原子性的寫入多個分區(qū)，即寫入到多個分區(qū)的消息要么全部成功，要么全部回滾。

這時候就需要使用事務(wù)，在producer端設(shè)置 transcational.id為一個指定字符串。

這樣冪等producer只能保證單分區(qū)上無重復(fù)消息；事務(wù)可以保證多分區(qū)寫入消息的完整性。

這樣producer端實現(xiàn)了exactly once，那么consumer端呢？

consumer端由于可能無法消費事務(wù)中所有消息，并且消息可能被刪除，所以事務(wù)并不能解決consumer端exactly once的問題，我們可能還是需要自己處理這方面的邏輯。比如自己管理offset的提交，不要自動提交，也是可以實現(xiàn)exactly once的。

還有一個選擇就是使用kafka自己的流處理引擎，也就是Kafka Streams，

設(shè)置processing.guarantee=exactly_once，就可以輕松實現(xiàn)exactly once了。