Kafka如果丟了消息，怎么處理的?
作者 |?Dong GuoChao
Kafka存在丟消息的問題，消息丟失會(huì)發(fā)生在Broker，Producer和Consumer三種。
Broker
Broker丟失消息是由于Kafka本身的原因造成的，kafka為了得到更高的性能和吞吐量，將數(shù)據(jù)異步批量的存儲(chǔ)在磁盤中。消息的刷盤過程，為了提高性能，減少刷盤次數(shù)，kafka采用了批量刷盤的做法。即，按照一定的消息量，和時(shí)間間隔進(jìn)行刷盤。這種機(jī)制也是由于linux操作系統(tǒng)決定的。將數(shù)據(jù)存儲(chǔ)到linux操作系統(tǒng)種，會(huì)先存儲(chǔ)到頁緩存（Page cache）中，按照時(shí)間或者其他條件進(jìn)行刷盤（從page cache到file），或者通過fsync命令強(qiáng)制刷盤。數(shù)據(jù)在page cache中時(shí)，如果系統(tǒng)掛掉，數(shù)據(jù)會(huì)丟失。
Broker在linux服務(wù)器上高速讀寫以及同步到Replica
上圖簡(jiǎn)述了broker寫數(shù)據(jù)以及同步的一個(gè)過程。broker寫數(shù)據(jù)只寫到PageCache中，而pageCache位于內(nèi)存。這部分?jǐn)?shù)據(jù)在斷電后是會(huì)丟失的。pageCache的數(shù)據(jù)通過linux的flusher程序進(jìn)行刷盤。刷盤觸發(fā)條件有三：
主動(dòng)調(diào)用sync或fsync函數(shù) 可用內(nèi)存低于閥值 dirty data時(shí)間達(dá)到閥值。dirty是pagecache的一個(gè)標(biāo)識(shí)位，當(dāng)有數(shù)據(jù)寫入到pageCache時(shí)，pagecache被標(biāo)注為dirty，數(shù)據(jù)刷盤以后，dirty標(biāo)志清除。
Broker配置刷盤機(jī)制，是通過調(diào)用fsync函數(shù)接管了刷盤動(dòng)作。從單個(gè)Broker來看，pageCache的數(shù)據(jù)會(huì)丟失。
Kafka沒有提供同步刷盤的方式。同步刷盤在RocketMQ中有實(shí)現(xiàn)，實(shí)現(xiàn)原理是將異步刷盤的流程進(jìn)行阻塞，等待響應(yīng)，類似ajax的callback或者是java的future。下面是一段rocketmq的源碼。
GroupCommitRequest?request?=?new?GroupCommitRequest(result.getWroteOffset()?+?result.getWroteBytes());
service.putRequest(request);
boolean?flushOK?=?request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());?//?刷盤
也就是說，理論上，要完全讓kafka保證單個(gè)broker不丟失消息是做不到的，只能通過調(diào)整刷盤機(jī)制的參數(shù)緩解該情況。比如，減少刷盤間隔，減少刷盤數(shù)據(jù)量大小。時(shí)間越短，性能越差，可靠性越好（盡可能可靠）。這是一個(gè)選擇題。
為了解決該問題，kafka通過producer和broker協(xié)同處理單個(gè)broker丟失參數(shù)的情況。一旦producer發(fā)現(xiàn)broker消息丟失，即可自動(dòng)進(jìn)行retry。除非retry次數(shù)超過閥值（可配置），消息才會(huì)丟失。此時(shí)需要生產(chǎn)者客戶端手動(dòng)處理該情況。那么producer是如何檢測(cè)到數(shù)據(jù)丟失的呢？是通過ack機(jī)制，類似于http的三次握手的方式。
The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are allowed:?
acks=0?If set to zero then the producer will not wait for any acknowledgment from the server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the?retries?configuration will not take effect (as the client won’t generally know of any failures). The offset given back for each record will always be set to -1.?acks=1?This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost.?acks=allThis means the leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting.http://kafka.apache.org/20/documentation.html
以上的引用是kafka官方對(duì)于參數(shù)acks的解釋（在老版本中，該參數(shù)是request.required.acks）。
acks=0，producer不等待broker的響應(yīng)，效率最高，但是消息很可能會(huì)丟。
acks=1，leader broker收到消息后，不等待其他follower的響應(yīng)，即返回ack。也可以理解為ack數(shù)為1。此時(shí)，如果follower還沒有收到leader同步的消息leader就掛了，那么消息會(huì)丟失。按照上圖中的例子，如果leader收到消息，成功寫入PageCache后，會(huì)返回ack，此時(shí)producer認(rèn)為消息發(fā)送成功。但此時(shí)，按照上圖，數(shù)據(jù)還沒有被同步到follower。如果此時(shí)leader斷電，數(shù)據(jù)會(huì)丟失。
acks=-1，leader broker收到消息后，掛起，等待所有ISR列表中的follower返回結(jié)果后，再返回ack。-1等效與 all 。這種配置下，只有l(wèi)eader寫入數(shù)據(jù)到pagecache是不會(huì)返回ack的，還需要所有的ISR返回“成功”才會(huì)觸發(fā)ack。如果此時(shí)斷電，producer可以知道消息沒有被發(fā)送成功，將會(huì)重新發(fā)送。如果在follower收到數(shù)據(jù)以后，成功返回ack，leader斷電，數(shù)據(jù)將存在于原來的follower中。在重新選舉以后，新的leader會(huì)持有該部分?jǐn)?shù)據(jù)。數(shù)據(jù)從leader同步到follower，需要2步：
數(shù)據(jù)從pageCache被刷盤到disk。因?yàn)橹挥衐isk中的數(shù)據(jù)才能被同步到replica。 數(shù)據(jù)同步到replica，并且replica成功將數(shù)據(jù)寫入PageCache。在producer得到ack后，哪怕是所有機(jī)器都停電，數(shù)據(jù)也至少會(huì)存在于leader的磁盤內(nèi)。
上面第三點(diǎn)提到了ISR的列表的follower，需要配合另一個(gè)參數(shù)才能更好的保證ack的有效性。ISR是Broker維護(hù)的一個(gè)“可靠的follower列表”，in-sync Replica列表，broker的配置包含一個(gè)參數(shù)：min.insync.replicas。該參數(shù)表示ISR中最少的副本數(shù)。如果不設(shè)置該值，ISR中的follower列表可能為空。此時(shí)相當(dāng)于acks=1。
如上圖中：
acks=0，總耗時(shí)f(t) = f(1)。 acks=1，總耗時(shí)f(t) = f(1) + f(2)。 acks=-1，總耗時(shí)f(t) = f(1) + max( f(A) , f(B) ) + f(2)。
性能依次遞減，可靠性依次升高。
Producer
Producer丟失消息，發(fā)生在生產(chǎn)者客戶端。
為了提升效率，減少IO，producer在發(fā)送數(shù)據(jù)時(shí)可以將多個(gè)請(qǐng)求進(jìn)行合并后發(fā)送。被合并的請(qǐng)求咋發(fā)送一線緩存在本地buffer中。緩存的方式和前文提到的刷盤類似，producer可以將請(qǐng)求打包成“塊”或者按照時(shí)間間隔，將buffer中的數(shù)據(jù)發(fā)出。通過buffer我們可以將生產(chǎn)者改造為異步的方式，而這可以提升我們的發(fā)送效率。
但是，buffer中的數(shù)據(jù)就是危險(xiǎn)的。在正常情況下，客戶端的異步調(diào)用可以通過callback來處理消息發(fā)送失敗或者超時(shí)的情況，但是，一旦producer被非法的停止了，那么buffer中的數(shù)據(jù)將丟失，broker將無法收到該部分?jǐn)?shù)據(jù)。又或者，當(dāng)Producer客戶端內(nèi)存不夠時(shí)，如果采取的策略是丟棄消息（另一種策略是block阻塞），消息也會(huì)被丟失。抑或，消息產(chǎn)生（異步產(chǎn)生）過快，導(dǎo)致掛起線程過多，內(nèi)存不足，導(dǎo)致程序崩潰，消息丟失。
producer采取批量發(fā)送的示意圖
異步發(fā)送消息生產(chǎn)速度過快的示意圖
根據(jù)上圖，可以想到幾個(gè)解決的思路：
異步發(fā)送消息改為同步發(fā)送消。或者service產(chǎn)生消息時(shí)，使用阻塞的線程池，并且線程數(shù)有一定上限。整體思路是控制消息產(chǎn)生速度。 擴(kuò)大Buffer的容量配置。這種方式可以緩解該情況的出現(xiàn)，但不能杜絕。 service不直接將消息發(fā)送到buffer（內(nèi)存），而是將消息寫到本地的磁盤中（數(shù)據(jù)庫或者文件），由另一個(gè)（或少量）生產(chǎn)線程進(jìn)行消息發(fā)送。相當(dāng)于是在buffer和service之間又加了一層空間更加富裕的緩沖層。
Consumer
Consumer消費(fèi)消息有下面幾個(gè)步驟：
接收消息 處理消息 反饋“處理完畢”（commited）
Consumer的消費(fèi)方式主要分為兩種：
自動(dòng)提交offset，Automatic Offset Committing 手動(dòng)提交offset，Manual Offset Control
Consumer自動(dòng)提交的機(jī)制是根據(jù)一定的時(shí)間間隔，將收到的消息進(jìn)行commit。commit過程和消費(fèi)消息的過程是異步的。也就是說，可能存在消費(fèi)過程未成功（比如拋出異常），commit消息已經(jīng)提交了。此時(shí)消息就丟失了。
Properties?props?=?new?Properties();
props.put("bootstrap.servers",?"localhost:9092");
props.put("group.id",?"test");
//?自動(dòng)提交開關(guān)
props.put("enable.auto.commit",?"true");
//?自動(dòng)提交的時(shí)間間隔，此處是1s
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");
KafkaConsumer?consumer?=?new?KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo",?"bar"));
while?(true)?{
????????//?調(diào)用poll后，1000ms后，消息狀態(tài)會(huì)被改為?committed
?ConsumerRecords?records?=?consumer.poll(100);
?for?(ConsumerRecord?record?:?records)
??insertIntoDB(record);?//?將消息入庫，時(shí)間可能會(huì)超過1000ms
}
上面的示例是自動(dòng)提交的例子。如果此時(shí)，insertIntoDB(record)發(fā)生異常，消息將會(huì)出現(xiàn)丟失。接下來是手動(dòng)提交的例子：
Properties?props?=?new?Properties();
props.put("bootstrap.servers",?"localhost:9092");
props.put("group.id",?"test");
//?關(guān)閉自動(dòng)提交，改為手動(dòng)提交
props.put("enable.auto.commit",?"false");
props.put("key.deserializer",?"org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer",?"org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer?consumer?=?new?KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo",?"bar"));
final?int?minBatchSize?=?200;
List>?buffer?=?new?ArrayList<>();
while?(true)?{
????????//?調(diào)用poll后，不會(huì)進(jìn)行auto?commit
???ConsumerRecords?records?=?consumer.poll(100);
???for?(ConsumerRecord?record?:?records)?{
????buffer.add(record);
???}
???if?(buffer.size()?>=?minBatchSize)?{
????insertIntoDb(buffer);
????????????????//?所有消息消費(fèi)完畢以后，才進(jìn)行commit操作
????consumer.commitSync();
????buffer.clear();
???}
?}
將提交類型改為手動(dòng)以后，可以保證消息“至少被消費(fèi)一次”(at least once)。但此時(shí)可能出現(xiàn)重復(fù)消費(fèi)的情況，重復(fù)消費(fèi)不屬于本篇討論范圍。
上面兩個(gè)例子，是直接使用Consumer的High level API，客戶端對(duì)于offset等控制是透明的。也可以采用Low level API的方式，手動(dòng)控制offset，也可以保證消息不丟，不過會(huì)更加復(fù)雜。
??try?{
?????while(running)?{
?????????ConsumerRecords?records?=?consumer.poll(Long.MAX_VALUE);
?????????for?(TopicPartition?partition?:?records.partitions())?{
?????????????List>?partitionRecords?=?records.records(partition);
?????????????for?(ConsumerRecord?record?:?partitionRecords)?{
?????????????????System.out.println(record.offset()?+?":?"?+?record.value());
?????????????}
?????????????long?lastOffset?=?partitionRecords.get(partitionRecords.size()?-?1).offset();
?????????????//?精確控制offset
?????????????consumer.commitSync(Collections.singletonMap(partition,?new?OffsetAndMetadata(lastOffset?+?1)));
?????????}
?????}
??}?finally?{
????consumer.close();
??} 推薦閱讀