面試題來源：

《大數據面試題 V4.0》

大數據面試題V3.0，523道題，679頁，46w字

可回答：Kafka如何保證生產者不丟失數據，消費者不丟失數據？

參考答案：

存在數據丟失的幾種情況

• 使用同步模式的時候，有3種狀態(tài)保證消息被安全生產，在配置為1（只保證寫入leader成功）的話，如果剛好leader partition掛了，數據就會丟失。

• 還有一種情況可能會丟失消息，就是使用異步模式的時候，當緩沖區(qū)滿了，如果配置為0（還沒有收到確認的情況下，緩沖池一滿，就清空緩沖池里的消息），數據就會被立即丟棄掉。

避免方法的一些概述

1、在數據生產時避免數據丟失的方法

只要能避免上述兩種情況，那么就可以保證消息不會被丟失。

1）在同步模式的時候，確認機制設置為-1，也就是讓消息寫入leader和所有的副本。

2）在異步模式下，如果消息發(fā)出去了，但還沒有收到確認的時候，緩沖池滿了，在配置文件中設置成不限制阻塞超時的時間，也就說讓生產端一直阻塞，這樣也能保證數據不會丟失。

在數據消費時，避免數據丟失的方法：確認數據被完成處理之后，再更新offset值。低級API中需要手動控制offset值。

消息隊列的問題都要從源頭找問題，就是生產者是否有問題。

討論一種情況，如果數據發(fā)送成功，但是接受response的時候丟失了，機器重啟之后就會重發(fā)。

重發(fā)很好解決，消費端增加去重表就能解決，但是如果生產者丟失了數據，問題就很麻煩了。

2、數據重復消費的情況，處理情況如下

1）去重：將消息的唯一標識保存到外部介質中，每次消費處理時判斷是否處理過；

2）不管：大數據場景中，報表系統(tǒng)或者日志信息丟失幾條都無所謂，不會影響最終的統(tǒng)計分析結。

Kafka到底會不會丟數據(data loss)? 通常不會，但有些情況下的確有可能會發(fā)生。下面的參數配置及Best practice列表可以較好地保證數據的持久性(當然是trade-off，犧牲了吞吐量)。

如果想要高吞吐量就要能容忍偶爾的失?。ㄖ匕l(fā)漏發(fā)無順序保證）。

?block.on.buffer.full = true
?acks = all
?retries = MAX_VALUE
?max.in.flight.requests.per.connection = 1
?使用KafkaProducer.send(record, callback)
?callback邏輯中顯式關閉producer：close(0) 
?unclean.leader.election.enable=false
?replication.factor = 3 
?min.insync.replicas = 2
?replication.factor > min.insync.replicas
?enable.auto.commit=false

消息處理完成之后再提交位移

給出列表之后，我們從兩個方面來探討一下數據為什么會丟失：

Producer端

新版本的Kafka替換了Scala版本的old producer，使用了由Java重寫的producer。新版本的producer采用異步發(fā)送機制。KafkaProducer.send(ProducerRecord)方法僅僅是把這條消息放入一個緩存中(即RecordAccumulator，本質上使用了隊列來緩存記錄)，同時后臺的IO線程會不斷掃描該緩存區(qū)，將滿足條件的消息封裝到某個batch中然后發(fā)送出去。顯然，這個過程中就有一個數據丟失的窗口：若IO線程發(fā)送之前client端掛掉了，累積在accumulator中的數據的確有可能會丟失。

Producer的另一個問題是消息的亂序問題。假設客戶端代碼依次執(zhí)行下面的語句將兩條消息發(fā)到相同的分區(qū)

?producer.send(record1);
?producer.send(record2);

如果此時由于某些原因(比如瞬時的網絡抖動)導致record1沒有成功發(fā)送，同時Kafka又配置了重試機制和max.in.flight.requests.per.connection大于1(默認值是5，本來就是大于1的)，那么重試record1成功后，record1在分區(qū)中就在record2之后，從而造成消息的亂序。很多某些要求強順序保證的場景是不允許出現(xiàn)這種情況的。發(fā)送之后重發(fā)就會丟失順序。

鑒于producer的這兩個問題，我們應該如何規(guī)避呢？？對于消息丟失的問題，很容易想到的一個方案就是：既然異步發(fā)送有可能丟失數據， 我改成同步發(fā)送總可以吧？比如這樣：

?producer.send(record).get();

這樣當然是可以的，但是性能會很差，不建議這樣使用。以下的配置清單應該能夠比較好地規(guī)避producer端數據丟失情況的發(fā)生：(特此說明一下，軟件配置的很多決策都是trade-off，下面的配置也不例外：應用了這些配置，你可能會發(fā)現(xiàn)你的producer/consumer 吞吐量會下降，這是正常的，因為你換取了更高的數據安全性)。

block.on.buffer.full = true 盡管該參數在0.9.0.0已經被標記為“deprecated”，但鑒于它的含義非常直觀，所以這里還是顯式設置它為true，使得producer將一直等待緩沖區(qū)直至其變?yōu)榭捎?。否則如果producer生產速度過快耗盡了緩沖區(qū)，producer將拋出異常。緩沖區(qū)滿了就阻塞在那，不要拋異常，也不要丟失數據。

acks=all 很好理解，所有follower都響應了才認為消息提交成功，即"committed"。

retries = MAX 無限重試，直到你意識到出現(xiàn)了問題。

max.in.flight.requests.per.connection = 1 限制客戶端在單個連接上能夠發(fā)送的未響應請求的個數。設置此值是1表示kafka broker在響應請求之前client不能再向同一個broker發(fā)送請求。注意：設置此參數是為了避免消息亂序。

使用KafkaProducer.send(record, callback)而不是send(record)方法 自定義回調邏輯處理消息發(fā)送失敗，比如記錄在日志中，用定時腳本掃描重處理。

callback邏輯中最好顯式關閉producer：close(0) 注意：設置此參數是為了避免消息亂序（僅僅因為一條消息發(fā)送沒收到反饋就關閉生產者，感覺代價很大）。

unclean.leader.election.enable=false 關閉unclean leader選舉，即不允許非ISR中的副本被選舉為leader，以避免數據丟失。 replication.factor >= 3 參考了Hadoop及業(yè)界通用的三備份原則。

min.insync.replicas > 1 消息至少要被寫入到這么多副本才算成功，也是提升數據持久性的一個參數。與acks配合使用保證replication.factor > min.insync.replicas 如果兩者相等，當一個副本掛掉了分區(qū)也就沒法正常工作了。通常設置replication.factor = min.insync.replicas + 1即可。

Consumer端

consumer端丟失消息的情形比較簡單：如果在消息處理完成前就提交了offset，那么就有可能造成數據的丟失。由于Kafka consumer默認是自動提交位移的，所以在后臺提交位移前一定要保證消息被正常處理了，因此不建議采用很重的處理邏輯，如果處理耗時很長，則建議把邏輯放到另一個線程中去做。為了避免數據丟失，現(xiàn)給出兩點建議：

• enable.auto.commit=false，關閉自動提交位移

• 在消息被完整處理之后再手動提交位移文章來源地址http://www.zghlxwxcb.cn/news/detail-605459.html

到了這里，關于大數據面試題：Kafka怎么保證數據不丟失，不重復？的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！