科技

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

ApacheKafka · 發表 2018-12-01 21:55:58

摘要：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。期待加入IOT時代最具戰鬥力的團隊。QQ郵箱地址：[email protected]，如有任何學術交流，可隨時聯絡。 ...

本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。期待加入IOT時代最具戰鬥力的團隊。QQ郵箱地址：[email protected]，如有任何學術交流，可隨時聯絡。

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 事務：跨分割槽原子寫入

事務：跨分割槽原子寫入

將允許一個生產者傳送一批到不同分割槽的訊息，這些訊息要麼全部對任何一個消費者可見，要麼對任何一個消費者都不可見。這個特性也允許你在一個事務中處理消費資料和提交消費偏移量，從而實現端到端的精確一次語義。

主要針對訊息經過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”消費者配置表示：

read_commited：除了讀取不屬於事務的訊息之外，還可以讀取事務提交後的訊息。
read_uncommited：按照偏移位置讀取所有訊息，而不用等事務提交。這個選項類似Kafka消費者的當前語義。
複製程式碼

為了使用事務，需要配置消費者使用正確的隔離等級。
使用新版生產者，並且將生產者的“transactional . id”配置項設定為某個唯一ID。需要此唯一ID來提供跨越應用程式重新啟動的事務狀態的連續性。

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

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

1.設定enable.auto.commit = false;
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;
}

}
複製程式碼

6總結

Kafka 0.11.0.0版本的逆天之作，都是在消費者EOS語義較弱，需要進一步增強。

秦凱新於深圳 201812012146

來源：稀土掘金