kafka精確一次語義EOS的原理深刻剖析-kafka 商業環境實戰

時間 2019-11-16

標籤 kafka 精確一次語義 eos 原理深刻剖析商業環境實戰欄目 Kafka 简体版

原文原文鏈接

本套技術專欄是做者（秦凱新）平時工做的總結和昇華，經過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和集羣環境容量規劃等內容，請持續關注本套博客。期待加入IOT時代最具戰鬥力的團隊。QQ郵箱地址：1120746959@qq.com，若有任何學術交流，可隨時聯繫。apache

1 Kafka 0.11.0.0版本的逆天之做

0.11.0.0版本以前默認提供at least once語義，想象這樣一種場景，分區的Leader副本所在的Broker成功的將消息寫入本地磁盤，而後broker將發送響應給producer，此時假設網絡出現故障致使該響應沒有發送成功。此種狀況下，Producer將認爲消息發送請求失敗，從而開啓重試機制。若此時網絡恢復正常，那麼同一條消息將會被寫入兩次。
基於上述案例：0.11.0.0版本提供冪等性：每一個分區中精確一次且有序
0.11.0.0版本提供事務：跨分區原子寫入機制。

2 故障類型

broker可能故障：Kafka是一個高可用、持久化的系統，每一條寫入一個分區的消息都會被持久化而且多副本備份（假設有n個副本）。因此，Kafka能夠容忍n-1個broker故障，意味着一個分區只要至少有一個broker可用，分區就可用。Kafka的副本協議保證了只要消息被成功寫入了主副本，它就會被複制到其餘全部的可用副本（ISR）。
producer到broker的RPC調用可能失敗：Kafka的持久性依賴於生產者接收broker的ack響應。沒有接收成功ack不表明生產請求自己失敗了。broker可能在寫入消息後，發送ack給生產者的時候掛了。甚至broker也可能在寫入消息前就掛了。因爲生產者沒有辦法知道錯誤是什麼形成的，因此它就只能認爲消息沒寫入成功，而且會重試發送。在一些狀況下，這會形成一樣的消息在Kafka分區日誌中重複，進而形成消費端屢次收到這條消息。
客戶端可能會故障：精確一次交付也必須考慮客戶端故障。可是咱們如何知道一個客戶端已經故障而不是暫時和brokers斷開，或者經歷一個程序短暫的暫停，區分永久性故障和臨時故障是很重要的，爲了正確性，broker應該丟棄僵住的生產這發送來的消息，一樣，也應該不向已經僵住的消費者發送消息。一旦一個新的客戶端實例啓動，它應該可以從失敗的實例留下的任何狀態中恢復，從一個安全點開始處理。這意味着，消費的偏移量必須始終與生產的輸出保持同步。

3 Producer冪等性處理機制

若是出現致使生產者重試的錯誤，一樣的消息，仍由一樣的生產者發送屢次，將只被寫到kafka broker的日誌中一次。對於單個分區，冪等生產者不會由於生產者或broker故障而發送多條重複消息。
kafka保存序列號僅僅須要幾個額外的字段，所以這種機制的開銷很是低。
除了序列號，kafka會爲每一個Producer實例分配一個Producer id（PID）,每一條消息都會有序列號，並嚴格遞增順序。若發送的消息的序列號小於或者等於broker端保存的序列號，那麼broker會拒絕這條消息的寫入操做。
注意的是：當前的設計只能保證單個producer實例的EOS語義，沒法實現多個Producer實例一塊提供EOS語義。
想要開啓這個特性，得到每一個分區內的精確一次語義，也就是說沒有重複，沒有丟失，而且有序的語義，只須要設置producer配置中的」enable.idempotence=true」。

4 事務：跨分區原子寫入

事務：跨分區原子寫入bootstrap

將容許一個生產者發送一批到不一樣分區的消息，這些消息要麼所有對任何一個消費者可見，要麼對任何一個消費者都不可見。這個特性也容許你在一個事務中處理消費數據和提交消費偏移量，從而實現端到端的精確一次語義。安全

主要針對消息通過Partioner分區器到多個分區的狀況。網絡

producer.initTransactions();
  try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
  } catch(ProducerFencedException e) {
    producer.close();
  } catch(KafkaException e) {
    producer.abortTransaction();
  }
複製代碼

5 消費端的事務支持

在消費者方面，有兩種選擇來讀取事務性消息，經過隔離等級「isolation.level」消費者配置表示：session

read_commited：除了讀取不屬於事務的消息以外，還能夠讀取事務提交後的消息。
  read_uncommited：按照偏移位置讀取全部消息，而不用等事務提交。這個選項相似Kafka消費者的當前語義。
複製代碼

爲了使用事務，須要配置消費者使用正確的隔離等級。ide
使用新版生產者，而且將生產者的「transactional . id」配置項設置爲某個惟一ID。須要此惟一ID來提供跨越應用程序從新啓動的事務狀態的連續性。fetch

6 消費端精確到一次語義實現

消費端精確到一次語義實現：consumer經過subscribe方法註冊到kafka，精確一次的語義要求必須手動管理offset，按照下述步驟進行設置：this

1.設置enable.auto.commit = false;spa
2.處理完消息以後不要手動提交offset，設計
3.經過subscribe方法將consumer註冊到某個特定topic，
4.實現ConsumerRebalanceListener接口和consumer.seek(topicPartition,offset)方法（讀取特定topic和partition的offset）

5.將offset和消息一塊存儲，確保原子性，推薦使用事務機制。

public class ExactlyOnceDynamicConsumer {

  private static OffsetManager offsetManager = new OffsetManager("storage2");

  public static void main(String[] str) throws InterruptedException {

      System.out.println("Starting ManualOffsetGuaranteedExactlyOnceReadingDynamicallyBalancedPartitionConsumer ...");

      readMessages();

  }



  private static void readMessages() throws InterruptedException {

      KafkaConsumer<String, String> consumer = createConsumer();

      // Manually controlling offset but register consumer to topics to get dynamically assigned partitions.
      // Inside MyConsumerRebalancerListener use consumer.seek(topicPartition,offset) to control offset

      consumer.subscribe(Arrays.asList("normal-topic"), new MyConsumerRebalancerListener(consumer));

      processRecords(consumer);
  }


  private static KafkaConsumer<String, String> createConsumer() {
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      String consumeGroup = "cg3";

      props.put("group.id", consumeGroup);

      props.put("enable.auto.commit", "false");
      props.put("heartbeat.interval.ms", "2000");
      props.put("session.timeout.ms", "6001");

      * Control maximum data on each poll, make sure this value is bigger than the maximum single record size
      props.put("max.partition.fetch.bytes", "140");

      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      return new KafkaConsumer<String, String>(props);
  }

  private static void processRecords(KafkaConsumer<String, String> consumer) {

      while (true) {

          ConsumerRecords<String, String> records = consumer.poll(100);

          for (ConsumerRecord<String, String> record : records) {

              System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
              offsetManager.saveOffsetInExternalStore(record.topic(), record.partition(), record.offset());

          }
      }
  }
複製代碼

}

public class MyConsumerRebalancerListener implements org.apache.kafka.clients.consumer.ConsumerRebalanceListener {

  private OffsetManager offsetManager = new OffsetManager("storage2");
  private Consumer<String, String> consumer;

  public MyConsumerRebalancerListener(Consumer<String, String> consumer) {
      this.consumer = consumer;
  }

  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

      for (TopicPartition partition : partitions) {

          offsetManager.saveOffsetInExternalStore(partition.topic(), partition.partition(), consumer.position(partition));
      }
  }

  public void onPartitionsAssigned(Collection<TopicPartition> partitions) {


      for (TopicPartition partition : partitions) {
          consumer.seek(partition, offsetManager.readOffsetFromExternalStore(partition.topic(), partition.partition()));
      }
  }
複製代碼

}

public class OffsetManager {
    private String storagePrefix;
    public OffsetManager(String storagePrefix) {
        this.storagePrefix = storagePrefix;
    }

    void saveOffsetInExternalStore(String topic, int partition, long offset) {

        try {

            FileWriter writer = new FileWriter(storageName(topic, partition), false);

            BufferedWriter bufferedWriter = new BufferedWriter(writer);
            bufferedWriter.write(offset + "");
            bufferedWriter.flush();
            bufferedWriter.close();

        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    long readOffsetFromExternalStore(String topic, int partition) {

        try {

            Stream<String> stream = Files.lines(Paths.get(storageName(topic, partition)));

            return Long.parseLong(stream.collect(Collectors.toList()).get(0)) + 1;

        } catch (Exception e) {
            e.printStackTrace();
        }

        return 0;
    }

    private String storageName(String topic, int partition) {
        return storagePrefix + "-" + topic + "-" + partition;
    }

}
複製代碼