Kafka 會丟失信息嗎？

許多開發(fā)人員普遍認(rèn)為，Kafka 的設(shè)計本身就能保證不會丟失消息。然而，Kafka 架構(gòu)和配置的細(xì)微差別會導(dǎo)致消息的丟失。我們需要了解它如何以及何時可能丟失消息，并防止此類情況的發(fā)生。

下圖顯示了消息在 Kafka 的生命周期中可能丟失的場景。

圖片

01 生產(chǎn)者（Producer）

當(dāng)我們調(diào)用 producer.send() 發(fā)送消息時，消息不會直接發(fā)送到代理。

消息發(fā)送過程涉及兩個線程和一個隊列：

1. 應(yīng)用程序線程
2. 消息累加器
3. 發(fā)送線程（I/O 線程）

我們需要為生產(chǎn)者配置適當(dāng)?shù)?"acks "和 "retries"，以確保消息被發(fā)送到代理。

02 消息代理（Broker）

當(dāng)代理集群正常運行時，它不應(yīng)該丟失消息。但是，我們需要了解哪些極端情況可能會導(dǎo)致消息丟失：

1. 為了提高 I/O 吞吐量，消息通常會異步刷到磁盤上，因此如果實例在刷新之前宕機，消息就會丟失。
2. Kafka 集群中的副本需要正確配置，以保持?jǐn)?shù)據(jù)的有效副本。數(shù)據(jù)同步的確定性非常重要。

03 消費者（Consumer）

Kafka 提供了不同的提交消息的方式。自動提交可能會在實際處理記錄之前確認(rèn)對記錄的處理。當(dāng)消費者在處理過程中宕機時，有些記錄可能永遠(yuǎn)不會被處理。

一個好的做法是將同步提交和異步提交結(jié)合起來，在處理消息的循環(huán)中使用異步提交以提高吞吐量，在異常處理中使用同步提交以確保最后的偏移始終被提交。

下圖是這個方法的偽代碼：

try {
    while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

    for (ConsumerRecord<String, String> record : records) {
        // process records one by one
    }

    consumer.commitAsync();
    }
} catch (Exception e){
    // exception handling 
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }      
}