可靠的資料傳輸是系統的一個必要屬性，就像效能一樣，必須從一開始就設計到系統中。Apache Kafka在可靠的資料傳輸方面非常靈活，支援非常多的配置引數。

1. 可靠性保證

當我們討論可靠性時，通常會提到保證這個術語。最著名的可靠性保證ACID，它是關係型資料庫普遍支援的標準可靠性保證。理解Kafka提供的保證對於建立可靠的應用是至關重要的，Kafka能夠保證：

• 同一個分割槽訊息的順序保證。在同一分割槽裡使用相同的生產者，如果訊息B是在訊息A之後寫入的，那麼Kafka保證訊息B的offset會大於訊息A，並且消費者會先讀取訊息A。
• 只要至少有一個副本保持alive狀態，已經提交的訊息就不會丟失。
• 消費者只能讀取已經提交的訊息。

2. Broker配置

Broker有三個可以改變Kafka關於可靠訊息儲存行為的配置引數：

2.1 複製因子

default.replication.factor：自動建立topic時預設的複製因子，預設為1
replication.factor：在Kafka Streams client library配置，預設為1

複製因子N允許在N-1個brokers故障時仍能夠可靠地讀取和寫入資料。因此，設定更大的複製因子可以提高可用性和可靠性。在另一方面，對於複製因子N需要至少N個brokers，儲存N個數據副本，意味著需要N倍的磁碟空間，基本上就是使用更多的硬體換取更高的可用性。

那麼如何決定一個topic的複製因子？答案取決於topic的重要程度和預算。如果對於重啟一個broker（叢集的正常操作之一）導致特定的topic不可用時覺得OK，那麼複製因子1可能就足夠了。複製因子2表示可以在一個broker故障時仍然可用，但要注意的是，在一個broker故障的情況下也有可能導致叢集不穩定（大多數發生在舊版本）。因此，我們建議設定複製因子至少為3。

副本的放置位置也是非常重要的，預設地，Kafka將確保分割槽的每個副本都放到不同的broker。然而，在某些情況下，這還不夠安全。如果分割槽所有副本的brokers都放在同一機架上，而且機架頂部的交換機出現問題，那麼無論複製因子是多少，分割槽都不可用。為了防止機架的故障，建議把brokers放到多個機架上，並使用broker.rack配置引數為每個broker配置機架名字。如果配置了機架名字，Kafka將確保分割槽的副本分佈在多個機架上，以確保更高的可用性。

2.2 Unclean Leader選舉

如之前所述，當分割槽的leader不再可用時，其中一個同步副本會被選舉為新的leader。這種leader選舉是“clean”的，因為它保證已經提交的資料不會丟失。但如果除了剛剛變為不可用的leader之外沒有in-sync副本時，Kafka會怎樣處理？這種情況可能發生在以下兩種情況之一：

• 分割槽有3個副本，其中2個followers因故障變為不可用。在這種情況下，當生產者繼續寫入資料到leader時，所有訊息都會被認為已經提交（因為這個時候leader是唯一的同步副本）。如果leader也出現故障，而其中一個非同步副本重新啟動，那麼該非同步副本將變為分割槽唯一可用的副本。
• 分割槽有3個副本，但由於網路問題，其中2個followers在預設10秒內沒有同步到最新的訊息從而變為非同步副本。這個時候leader是唯一的同步副本，如果leader出現故障，2個followers不會再變為同步副本。

在這兩種情況下，我們要決定：

• 如果不允許非同步副本變成新的leader，相應的分割槽將一直保持不可用直到原來的leader重新恢復可用。
• 如果確實要允許非同步副本變成新的leader，這樣會丟失已經寫入到原來的leader但沒有同步到副本的資料，並且還會導致消費者的某些不一致。

總之，如果允許非同步副本變成新的leader就會面臨資料丟失和不一致的風險。如果不允許，將面臨較低的可用性。unclean.leader.election.enable設定預設為false。

2.3 最少同步副本數

在topic和broker的配置都是min.insync.replicas，預設是1。

正如上述提到的，有些情況下即使配置了三個副本，也可能會留下只有一個同步副本。如果此副本變得不可用，我們不得不在可用性和一致性之間進行取捨，這通常不是一個容易的選擇。如果希望確保已提交的資料寫入多個副本，需要增大最少同步副本數。例如一個topic有3個副本，把min.insync.replicas設定為2，那麼只有至少有2個副本完成同步資料時才算真正寫入到分割槽。當所有3個副本都同步時，一切都正常工作。如果其中1個副本變得不可用，也可以正常工作。但是，如果有2個副本不可用，brokers將不再接受寫入訊息的請求。生產者會收到NotEnoughReplicasException的異常，消費者可以繼續讀取存在的訊息，換句話說僅存的單個同步副本會變為只讀。為了從這種只讀的情況中恢復正常，必須使2個不可用的分割槽中的一個恢復可用（例如重啟broker）並等待它完成資料同步。

3. 相應的生產者配置

即使盡可能地以最可靠的方式配置brokers，但如果不把生產者也按相應可靠性配置，整個系統仍可能會意外丟失資料。以下是兩個示例場景：

• brokers配置3個副本，並禁用unclean leader選舉，生產者的acks設定為1。當傳送並已寫入訊息到leader但還沒有同步到副本時，該leader向生產者返回一條“Message was written successfully”的資訊，並在訊息同步到副本前發生故障。其它的副本仍被認為是同步的（在replica.lag.time.max.ms時間範圍內），其中一個副本被選舉為新的leader。由於之前提到的訊息沒有同步到副本，因此會丟失，但生產者的應用會認為已經寫入成功。整個系統還是處於一致的，因為沒有消費者能讀取那些訊息，但從生產者的角度來看，訊息是丟失的。
• brokers配置3個副本，並禁用unclean leader選舉，生產者的acks設定為all。當嘗試向leader傳送一條訊息，但該leader發生故障，新的leader正在選舉中，Kafka向生產者返回”Leader not Available“的資訊。此時，如果生產者沒有異常處理和重試機制，訊息會丟失。這個不是broker的可靠性問題，因為broker從未收到那些訊息；也不是一致性的問題，因為消費者也從未讀取到那些訊息。但是如果生產者沒有正確處理異常，它們可能會導致訊息丟失。

那麼我們如何避免丟失訊息呢？如示例所示，必須注意以下兩個重要事項：

• 根據可靠性需求使用正確的acks配置
• 程式碼端要正確處理異常

前文已經詳細介紹過生產者相關的配置，但下面讓我們再次複習一下重點：

3.1 確認配置-acks

• acks=0，訊息一旦通過網路傳送就會被認為已經發送成功。如果傳送的訊息不能被序列化或者網路出現故障，生產者仍然會收到錯誤資訊，但如果分割槽處於下線狀態或整個叢集離線，生產者是不會收到任何錯誤資訊。這意味著即使出現預期的clean leader選舉，生產者也會丟失訊息，因為當正在選舉新的leader時，生產者是不會知道原來的leader不可用。使用配置acks=0，執行的效能是非常快的（這就是為什麼許多benchmarks測試使用該配置的原因）。你可以獲得驚人的吞吐量和使用大部分的頻寬，但肯定會丟失一些訊息。
• acks=1，leader在收到訊息並寫入分割槽資料檔案（但不一定同步到磁碟）會返回確認給生產者。如果在寫入訊息時發生故障，也會返回錯誤資訊。這意味著在正常的leader選舉情況下，當正在選舉leader時，生產者會收到LeaderNotAvailableException，如果生產者正確處理該異常，重新發送訊息，那麼訊息是不會丟失的。如果leader宕機並且訊息沒有被同步到副本，該訊息仍然會丟失。
• acks=all，在返回確認或異常資訊之前，leader將一直等待直到訊息同步到所有副本。與broker的min.insync.replicas配置結合使用，可以控制同步到訊息的副本數量。這是最安全的配置-在訊息完全提交之前，生產者會一直嘗試傳送訊息。但這也是最慢的配置-生產者會等待所有副本完成訊息同步才能繼續傳送下一批訊息。通過使用非同步模式和傳送更大批次的訊息可以減輕這些影響，但此配置始終會降低吞吐量。

3.2 重試配置-retries

生產者處理異常有2種方式：生產者自動處理和客戶程式碼處理。

生產者可以自動處理broker返回的可重試的retriable異常。當生產者向broker傳送訊息時，broker返回成功或異常資訊。異常資訊有2類-重試後可以解決的和無法解決的異常。例如，如果broker返回異常資訊LEADER_NOT_AVAILABLE，生產者可以嘗試重新發送訊息，有可能新的broker已經選舉出來，重新發送的訊息將會寫入成功。這意味著LEADER_NOT_AVAILABLE是一個可重試的retriable異常。另一方面，如果broker返回異常資訊INVALID_CONFIG，重試傳送資訊是不能更改配置的，這就是不可重試的nonretriable異常。

一般來說，如果你的目標是永遠不會丟失訊息，那麼最好的方法是配置生產者為在遇到可重試異常時重試傳送訊息。因為像leader不可用或網路連線問題通常需要幾秒鐘才能恢復，如果配置生產者重試傳送訊息直到成功，你就不需要自己處理這些異常。那麼應該配置重試多少次呢？如果當捕捉到異常後你想重試很多次，那麼肯定需要配置更多的重試次數。如果可以丟棄訊息，或者把它寫到其它地方稍後處理，那麼可以配置少的重試次數或不需要重試。例如，Kafka的跨資料中心複製工具（MirrorMaker）預設配置為無休止地重試，retries = MAX_INT，因為作為一個高可靠性的複製工具，它永遠不應該丟棄訊息。

注意，重試傳送失敗的訊息有小的機率會導致寫入相同的訊息。例如，如果網路問題導致broker無法向生產者傳送確認資訊，但訊息實際已經寫入成功並複製到副本，這種情況下生產者因為沒有收到確認資訊將會重發訊息，broker就會收到相同的訊息。重試和小心的異常處理可以保證每條訊息至少儲存一次，但在版本0.10.0不能保證僅儲存一次。有些應用程式會為每條訊息新增一個唯一標識，以便在讀取訊息時可以檢測重複性並去重。有些會把訊息設計為冪等性-即使相同的訊息被髮送2次也不會對正確性產生負面影響。例如，訊息“賬戶餘額為110元”，因為多次傳送它是不會改變結果。但訊息“向賬戶新增10元”不是冪等的，因為每次傳送它都會改變結果。

3.3 異常處理

使用生產者內建的重試機制是一種在不丟失訊息的情況下能夠正確處理大部分型別的異常的簡單方法，但作為開發人員，仍然必須處理其它型別的異常，包括：

• 不可重試的異常，例如關於訊息大小、授權等等的異常
• 訊息在被髮送給broker之前的異常，例如序列化異常
• 生產者重試次數到達上限的異常或者生產者的可用記憶體由於在重試時全部用來儲存訊息導致不足而發生的異常

前文已經介紹過如何在同步和非同步傳送訊息方法裡編寫異常處理程式碼，這裡不重複介紹。異常處理可以是記錄錯誤日誌，把訊息儲存在檔案系統以便後續處理，觸發一個callback函式呼叫另一個應用程式等等。需要注意的是，如果異常處理僅僅是重試傳送訊息，那麼最好使用生產者的重試功能。

4. 相應的消費者配置

至此我們已經介紹瞭如何在保證可靠性的前提下發送訊息，下面會介紹如何讀取訊息。

正如之前所述，訊息只有在提交後才能被消費者讀取，也就是說訊息被寫入到所有的同步副本，這意味著消費者可以讀取到一致的資料。消費者唯一要做的就是確保記錄哪些訊息是已經讀取，哪些沒有，這是不丟失訊息的關鍵。當從一個分割槽讀取訊息時，消費者是批量讀取的，然後記錄當前批次的最後一個offset，下次從記錄的offset開始讀取下一批訊息。這樣可以保證消費者以正確的順序讀取新訊息而不會遺漏任何訊息。

當一個消費者停止時，另一個消費者需要知道從哪裡接手工作-也就是上一個消費者在停止前處理的最後一個訊息的offset，這個消費者甚至可以是重啟後的原來那個消費者。這就是消費者需要提交它們的offsets的原因。消費者可能丟失訊息的主要原因是當提交offsets時，對應的訊息實際上是沒有處理完成的。這樣，當另一個消費者接手工作時，它將跳過那些訊息導致它們永遠不會被處理。這就是為什麼要特別注意何時以及怎樣提交offset的重要原因。

4.1 消費者的重要配置

為了使消費者有可靠性的行為，有4個重要的配置需要了解：

第一個是group.id，如果2個消費者有相同的group id並且訂閱相同的topic，那麼每個消費者將分配分割槽中的一個子集，因此只會讀取相應子集的訊息，但所有訊息會被整個組讀取。如果需要一個消費者讀取topic的每一條訊息，那麼需要為這個消費者設定唯一的group.id（該組只有唯一一個消費者）。

第二個相關配置是auto.offset.reset，此引數控制消費者在開始讀取一個沒有提交offset或該offset為非法的分割槽時如何重置offset。根據前文所述，可用的配置有earliest、latest(預設)、none。如果選擇earliest，消費者將從最開始讀取分割槽的所有訊息，這可能導致消費者重複處理大量的訊息，但可以保證丟失很少資料。如果選擇latest，消費者將會從最新的訊息開始讀取，這減少了訊息的重複處理，但幾乎肯定會導致資料丟失。

第三個相關配置是enable.auto.commit，是否自動週期性提交offset（預設為5秒）。如果消費者在poll迴圈中對訊息進行所有處理，那麼自動提交offset可以保證永遠不會提交未處理訊息的offset。自動提交offset的主要缺點是你無法控制可能需要處理的重複訊息數量（例如消費者在處理某些訊息後因某些原因停止了，但還沒有自動提交offset）。如果消費者把訊息發給另一個background執行緒來處理，可能會提交已經讀取但還沒有被處理訊息的offsets。

第四個相關配置與第三個相關，它是auto.commit.interval.ms。如果選擇自動提交offset，該配置允許你設定自動提交offset的間隔，預設是5秒。一般來說，更頻繁地提交會增加資源的使用，但會減少由於消費者停止可能導致的訊息重複數量。

4.2 消費者自己提交offsets

4.2.1 每次處理完訊息後都提交

如果在poll迴圈中對訊息進行所有處理，可以在每次poll迴圈結束時使用自動提交或自己提交offsets。

4.2.2 提交的頻率是決定效能和發生故障時有多少重複訊息的因數

即使在最簡單的情況下，在poll迴圈中對訊息進行所有處理，也可以選擇在每次迴圈多次提交或者每幾次迴圈才提交一次。提交offsets會使用一定的資源（類似於acks=all），所以提交頻率的選擇一切取決於你的需求。

4.2.3 確保知道提交的offset是哪個

在poll迴圈中提交offsets的常見陷阱是意外地提交最後讀取訊息的offset而不是最後處理訊息的offset。始終在處理訊息後提交offset是至關重要的，提交讀取訊息的offset會導致丟失訊息。

4.2.4 負載再均衡

在設計應用時，請記住消費者再均衡是會發生的，重要的是在分割槽被移除之前提交offsets並在分配新分割槽時清理任何維護的狀態。

4.2.5 消費者重試

在某些情況下，在呼叫poll迴圈處理訊息後，某些訊息沒有處理完畢需要稍後處理。例如，你可能嘗試把訊息寫入資料庫，但資料庫剛好不可用需要稍後重試。注意，與傳統的釋出/訂閱訊息系統不同的是，你提交的是offsets而不是訊息。這意味著如果你處理訊息#30失敗，但處理訊息#31成功，則不應該提交offset #31，否則會導致提交所有offset小於#31的訊息，也就是包括#30。以下是2種正確處理的選項：

選項1，當遇到可重試異常時，提交成功處理的最後一個訊息的offset。然後把處理失敗的訊息儲存在一個緩衝區中（以便下一個poll迴圈不會覆蓋它們）並繼續嘗試處理它們。你可以使用consumer.pause(Collection<TopicPartition> partitions)方法暫停從指定分割槽讀取訊息（這時如果呼叫poll方法將不會返回任何訊息），直到完成重試處理之前的訊息。

選項2，當遇到可重試異常時，把處理失敗的訊息寫到另外一個topic並繼續處理下一批訊息，然後使用另外一組消費者來重試處理這些訊息。

4.2.6 消費者維護狀態

在某些應用程式中，你需要在不同poll迴圈之間維護狀態。例如，計算移動的平均值，你需要在每次呼叫poll迴圈後更新該平均值。如果程序被重新啟動，你不僅需要從上一個offset開始讀取訊息，而且還需要恢復匹配之前計算的移動平均值，實現的其中一種方法可以在提交offset的同時把最新的累加值寫到另外一個topic中。這意味著當一個執行緒啟動時，它可以獲取最新的累加值並在上次停止的地方開始讀取訊息。但是，也有可能在把最新的累加值寫到另外一個topic後但在提交offsets前發生故障。一般來說，這是一個相當複雜的問題，建議嘗試使用類似Kafka Streams來解決這樣的問題，因為它為aggregation、joins、windows和其它複雜的分析提供了high level DSL-like APIs。

4.2.7 處理長時間的任務

有時處理訊息需要很長的時間，例如與一個可能阻塞或執行非常複雜計算的service進行互動。在某些版本的Kafka消費者中，你無法停止poll迴圈超過幾秒鐘。即使你不想處理更多的訊息，也必須繼續呼叫poll方法以便客戶端可以向broker傳送心跳。在這些情況下，一種常見的模式是把這些訊息放到一個執行緒池裡使用多執行緒來加快處理速度。一旦把訊息放到執行緒池裡，你可以暫停消費者從指定分割槽讀取訊息（呼叫consumer.pause方法，而且不會觸發負載再均衡）並繼續呼叫poll方法以便傳送心跳（這個時候不會讀取到任何訊息），直到訊息被處理完畢。

4.2.8 保證只發送一次

有些應用程式不僅需要至少傳送一次（保證沒有資料丟失），而且只需要傳送僅僅一次。雖然Kafka目前還沒有完全支援只發送一次，但消費者可以採用一些技巧保證每條訊息將會被寫入到外部系統僅一次。

最簡單且可能最常見的方法是將結果寫入一個支援唯一鍵的系統，這包括所有的key-value儲存，所有關係型資料庫，Elasticsearch以及其它資料儲存。你只需要建立一個唯一的key，可以使用topic、分割槽和offset的結合。當寫入相同的訊息時，後者會覆蓋前者因為它們有相同的key，這種模式稱為冪等寫。

另一種方法是將結果寫入一個支援事務的系統，關係型資料庫是最簡單的示例。例如在同一個事務裡把訊息和它們的offsets寫入到資料庫裡以便它們保持一致，當消費者重啟時，先從資料庫獲取最新的offset然後使用consumer.seek()方法從broker指定的offset讀取訊息。

5. 驗證系統的可靠性

5.1 驗證配置

從應用系統中單獨測試broker和客戶端配置是比較容易的，而且也建議這麼做，它有助於測試選擇的配置是否符合需求。Kafka包含兩個重要的工具來幫助驗證，org.apache.kafka.tools工具包包含VerifiableProducer和VerifiableConsumer類，它們可以作為命令列工具執行，也可以嵌入到自動化測試框架裡面。

使用VerifiableProducer生成一些訊息，其中包含從1到你選擇的值，然後設定acks、retries和生成訊息的速率。當執行時，它將根據接收到的acks為傳送給broker的每條訊息列印成功或錯誤資訊。VerifiableConsumer讀取訊息（通常是由VerifiableProducer生成）並按順序列印讀取到的訊息，它還會列印提交和負載再均衡的資訊。

也建議考慮執行以下測試：

• Leader選舉：如果把leader kill掉會發生什麼情況？生產者和消費者需要多久才能恢復正常工作？
• 控制器選舉：重啟控制器後系統需要多長時間才能恢復？
• 滾動重啟：可以逐個重啟brokers而不會丟失任何訊息嗎？
• Unclean leader選舉：如果逐個kill掉一個分割槽的所有副本（確保每個副本脫離同步）然後啟動一個非同步副本會發生什麼情況？為了恢復正常工作需要做些什麼？

然後選擇一個場景，啟動VerifiableProducer和VerifiableConsumer，並驗證該場景。例如，kill掉你正在寫入資料的leader。如果你期待短暫的停止，然後恢復正常而不會丟失訊息，請確保生產者生成的訊息數和消費者讀取的訊息數一致。另外，Apache Kafka的原始碼庫還包含一個擴充套件的測試套件。

5.2 驗證應用

一旦確定broker和客戶端配置符合需求，就可以測試整個應用程式是否也符合需求。這將驗證諸如自定義異常處理、offset的提交、負載再均衡等等的功能，特別是以下場景：

• 客戶端失去與伺服器的連線
• Leader選舉
• brokers的滾動重啟
• 消費者的滾動重啟
• 生產者的滾動重啟

5.3 監控系統的可靠性

測試應用程式很重要，但它不能取代持續監控生產系統以確保訊息按預期傳輸的需要。但除了監控叢集是否正常執行，還必須監控客戶端和訊息的傳輸。

首先，Kafka的Java客戶端包括允許監控客戶端狀態和事件的JMX指標。對於生產者來說，對可靠性最重要的2個指標是錯誤率和每條訊息的重試率。除了監控生產者的錯誤日誌，還要注意在傳送訊息時記錄的WARN級別日誌，例如“Got error produce response with correlation id 5689 on topic-partition [topic-1,3], retrying (two attempts left). Error:…”。

在消費者方面，最重要的指標是消費者滯後時間，這個指標顯示消費者與提交到broker分割槽最新訊息的距離。理想情況下，滯後總是0，也就是消費者將始終讀取最新的訊息。實際上，因為呼叫poll()方法返回批量訊息，然後消費者在讀取下一批訊息之前需要時間處理它們，所以滯後時間總是會有一點波動。重要的是確保消費者最終能夠追上而不是滯後時間變得更長。Burrow是linkedin開源的一個監控Apache Kafka的工具，可以用來監控消費者的滯後情況。

監控訊息的傳輸還意味著確保所有生成的訊息能夠被及時消費。為了實現這個，你需要知道訊息是何時生成的。從版本0.10.0開始，所有訊息都包含一個timestamp，顯示訊息的生成時間。為了確保在一個合理的時間內消費生成的所有訊息，你將需要應用程式來記錄生成的訊息數（通常是每秒訊息數）。消費者不僅需要記錄消費的訊息數（每秒訊息數），而且需要通過訊息的timestamp記錄從訊息的生成到被消費的滯後時間。然後你將需要一個系統來協調生產者和消費者的處理的每秒訊息數（確保沒有丟失訊息）並確保訊息被處理的滯後時間在一個合理的範圍內。為了實現更好地監控，你可以新增對關鍵topics的監控，但這些型別的end-to-end監控系統在實現起來可以說是極具挑戰和耗時的。

本文比較理論化，但希望你能把理論應用到實踐上面。

END O(∩_∩)O