Kafka叢集可靠性組態與最佳實踐

Kafka叢集的可靠性取決於Broker和客戶端的協同組態。Broker端需設定合理的ZooKeeper心跳間隔和副本延遲時間，並組態訊息持久化策略以平衡效能和可靠性。生產者端，acks設定決定訊息確認模式，retries和delivery.timeout.ms控制重試機制。消費者端，auto.offset.reset設定初始偏移量，enable.auto.commit和auto.commit.interval.ms控制偏移量提交策略。此外，還需考慮消費者端的錯誤處理和狀態維護，例如使用重試主題或跨輪詢呼叫維護狀態。最後，透過滾動重啟、故障注入測試以及生產環境監控，驗證並保障Kafka系統的可靠性。

Kafka 叢集可靠性組態與生產者設定

Kafka 叢集的可靠性對於確保資料不遺失至關重要。除了正確組態 Kafka broker 外，生產者的設定也扮演著關鍵角色。本文將探討 Kafka 的可靠性組態和生產者設定，以確保資料傳輸的安全性和一致性。

Kafka Broker 組態

Kafka 的 broker 組態對於叢集的穩定性和可靠性有直接影響。以下是一些重要的組態引數：

1. zookeeper.session.timeout.ms：此引數控制 Kafka broker 與 ZooKeeper 之間的心跳間隔。如果 broker 在此時間內未向 ZooKeeper 傳送心跳，則會被視為死亡並從叢集中移除。預設值為 18 秒。
2. replica.lag.time.max.ms：此引數控制副本與 leader 之間的最大延遲時間。如果副本在此時間內未從 leader 取得最新訊息，則會被視為不同步。預設值為 30 秒。

持久化到磁碟

Kafka 預設情況下不會將訊息立即寫入磁碟，而是依賴 Linux page cache 來快取訊息。這種設計可以提高效能，但也可能導致資料遺失。為了提高可靠性，可以組態 broker 將訊息更頻繁地寫入磁碟。

1. flush.messages：此引數控制在將訊息寫入磁碟之前可以接收的最大訊息數量。
2. flush.ms：此引數控制將訊息寫入磁碟的最大時間間隔。

生產者設定

生產者的設定對於確保資料傳輸的可靠性至關重要。以下是一些重要的設定：

1. acks：此引數控制生產者在傳送訊息時所需的確認模式。有三種模式：
   • acks=0：生產者不需要任何確認即可繼續傳送訊息。
   • acks=1：leader 需要確認收到訊息，但不需要等待副本確認。
   • acks=all：leader 需要等待所有副本確認收到訊息。

程式碼範例

// 設定 acks=all 以確保資料可靠性
Properties props = new Properties();
props.put("acks", "all");
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my_topic", "key", "value");
producer.send(record);

內容解密：

1. 設定 acks=all 以確保資料可靠性，意味著 leader 需要等待所有副本確認收到訊息。
2. 使用 Properties 物件來設定生產者的屬性，例如 bootstrap.servers、key.serializer 和 value.serializer。
3. 建立 KafkaProducer 物件並傳送 ProducerRecord 到指定的 topic。

Kafka 系統可靠性探討：生產者與消費者組態

在建構可靠的 Kafka 系統時，生產者（Producer）與消費者（Consumer）的組態至關重要。生產者的主要任務是將訊息傳送到 Kafka 叢集，而消費者的任務則是從 Kafka 叢集中讀取訊息。為了確保系統的可靠性，我們需要深入瞭解生產者和消費者的組態和錯誤處理機制。

生產者組態與可靠性

生產者的組態對於確保訊息的可靠傳送至關重要。其中一個重要的組態引數是 acks，它決定了生產者在認為訊息已成功傳送之前需要等待多少個副本的確認。

acks 引數詳解

• acks=0：生產者不會等待任何確認，直接傳送訊息。這種方式速度最快，但可靠性最低。
• acks=1：長官者（Leader）接收到訊息後即確認。這種方式在保證一定可靠性的同時，也保持了較好的效能。
• acks=all：長官者會等待所有同步副本（In-Sync Replicas）都接收到訊息後才確認。這是最安全的選項，但同時也是延遲最高的。

生產者重試機制

生產者內建了重試機制，用於處理可重試的錯誤，例如 LEADER_NOT_AVAILABLE。當遇到這種錯誤時，生產者會自動重試傳送訊息。為了避免訊息遺失，建議將重試次數設定為最大值（MAX_INT），並使用 delivery.timeout.ms 來控制最大等待時間。

// 設定重試次數為最大值
props.put("retries", Integer.MAX_VALUE);
// 設定最大等待時間為60秒
props.put("delivery.timeout.ms", 60000);

#### 內容解密：

此設定讓生產者在遇到可重試錯誤時，能夠在指定的時間內持續重試，直到成功或超時。這種方式能夠有效地提高訊息傳送的可靠性。

消費者組態與可靠性

消費者的任務是從 Kafka 叢集中讀取訊息，並確保訊息被正確處理。為了避免訊息遺失，消費者需要定期提交偏移量（Offsets）。

偏移量提交

消費者需要定期提交已處理訊息的偏移量，以確保在發生故障或重新平衡時，能夠從正確的位置繼續處理訊息。

// 自動提交偏移量
props.put("enable.auto.commit", "true");
// 設定自動提交間隔為5秒
props.put("auto.commit.interval.ms", "5000");

#### 內容解密：

自動提交偏移量的設定使得消費者能夠定時提交已處理的偏移量，從而避免因消費者故障導致的訊息遺失。然而，需要注意的是，如果消費者在處理訊息後但尚未提交偏移量之前發生故障，則可能會導致訊息被重複處理。

手動提交偏移量

為了更精確地控制偏移量的提交，可以採用手動提交的方式。

// 停用自動提交
props.put("enable.auto.commit", "false");

// 手動提交偏移量
consumer.commitSync();

#### 內容解密：

手動提交偏移量允許開發者在訊息處理完成後再提交偏移量，從而確保不會因為消費者故障而遺失訊息。然而，這種方式需要開發者自行管理偏移量的提交邏輯，增加了複雜度。

此圖示說明瞭生產者將訊息傳送到 Kafka 叢集，然後由消費者讀取並處理，最後提交偏移量到 Kafka 的流程。

Kafka 的可靠性和靈活性使其成為現代分散式系統中不可或缺的一部分。透過深入瞭解其內部機制和正確組態，我們可以充分發揮其潛力，建構出高效、可靠的資料處理系統。

組態消費者實作可靠處理的關鍵因素

在第四章中，我們詳細探討了消費者 API，並介紹了多種提交偏移量的方式。本章將重點討論相關的重要考量與選擇，但細節部分仍需參照第四章的內容。

可靠處理的重要消費者組態屬性

有四個消費者組態屬性對於理解如何組態消費者以實作所需的可靠性行為至關重要。

首先是 group.id，正如第四章所詳細解釋的那樣。如果兩個消費者擁有相同的群組 ID 並訂閱相同的主題，每個消費者將被分配到該主題的一個子集，因此只會個別讀取部分訊息（但整個群組將讀取所有訊息）。如果需要某個消費者能夠單獨讀取它訂閱的所有主題中的每條訊息，那麼它就需要一個獨特的 group.id。

第二個相關的組態是 auto.offset.reset。此引數控制著當沒有提交偏移量時（例如，消費者首次啟動時）或當消費者請求的偏移量在代理中不存在時的行為（第四章解釋了這種情況如何發生）。只有兩個選項可供選擇：如果選擇 earliest，則當消費者沒有有效的偏移量時，它將從分割槽的起始位置開始讀取。這可能會導致消費者處理大量重複的訊息，但它保證了最小化資料丟失。如果選擇 latest，則消費者將從分割槽的末尾開始讀取，這樣可以最小化重複處理，但幾乎肯定會導致某些訊息被遺漏。

第三個相關組態是 enable.auto.commit。這是一個重要的決定：我們是要讓消費者根據排程自動提交偏移量，還是在程式碼中手動提交偏移量？自動提交偏移量的主要好處是，在使用消費者的應用程式中，這減少了一件需要擔心的事。當我們在消費者輪詢迴圈內處理所有消費到的記錄時，自動偏移量提交保證了我們永遠不會意外地提交尚未處理的偏移量。自動偏移量提交的主要缺點是，我們無法控制應用程式可能處理的重複記錄數量，因為在處理某些記錄後但在自動提交啟動之前，應用程式可能會停止。當應用程式具有更複雜的處理過程，例如將記錄傳遞給另一個執行緒在後台處理時，則必須使用手動偏移量提交，因為自動提交可能會提交消費者已讀取但尚未處理的記錄的偏移量。

第四個相關組態 auto.commit.interval.ms 與第三個組態相關聯。如果我們選擇自動提交偏移量，此組態允許我們設定提交的頻率。預設值是每五秒。一般來說，更頻繁地提交會增加開銷，但會減少消費者停止時可能出現的重複數量。

明確在消費者中提交偏移量

如果我們決定需要更多的控制權並選擇手動提交偏移量，那麼我們需要關注正確性和效能影響。

總是在訊息被處理後提交偏移量

如果我們在輪詢迴圈內進行所有處理，並且不在輪詢迴圈之間保持狀態（例如，用於聚合），這應該很容易實作。我們可以使用自動提交組態，在輪詢迴圈結束時提交偏移量，或者在迴圈內以平衡開銷和重複處理需求的頻率提交偏移量。如果涉及額外的執行緒或狀態處理，這就會變得更加複雜，尤其是因為消費者物件不是執行緒安全的。在第四章中，我們討論瞭如何做到這一點，並提供了更多範例的參考。

提交頻率是在當機事件中效能和重複數量之間的權衡

即使在最簡單的情況下，我們在輪詢迴圈內進行所有處理，並且不在輪詢迴圈之間保持狀態，我們也可以選擇在迴圈內多次提交，或者選擇每隔幾次迴圈才提交一次。提交具有顯著的效能開銷。它類別似於使用 acks=all 的生產者，但單個消費者群組的所有偏移量提交都會被生產到同一個代理，這可能會導致過載。提交頻率需要在效能需求和避免重複之間取得平衡。在非常低吞吐量的主題上，才應該考慮每次訊息都提交。

在正確的時間提交正確的偏移量

在輪詢迴圈中間提交時，一個常見的陷阱是意外地提交了輪詢時讀取的最後一個偏移量，而不是最後處理的偏移量之後的偏移量。請記住，總是在訊息被處理後提交偏移量對於避免遺漏訊息至關重要。第四章中有範例展示瞭如何正確地做到這一點。

重新平衡

在設計應用程式時，我們需要記住消費者重新平衡將會發生，並且需要正確地處理它們。第四章包含了一些範例。這通常涉及在分割槽被復原之前提交偏移量，以及在應用程式被分配新的分割槽時清理應用程式維護的任何狀態。

消費者可能需要重試

在某些情況下，在呼叫輪詢並處理記錄之後，某些記錄可能沒有被完全處理，需要稍後再處理。例如，我們可能嘗試將記錄從 Kafka 寫入資料函式庫，但發現資料函式庫在那一刻不可用，需要稍後重試。

程式碼範例與解析

from kafka import KafkaConsumer

# 建立 KafkaConsumer 例項
consumer = KafkaConsumer('my_topic',
                         group_id='my_group',
                         bootstrap_servers=['localhost:9092'],
                         auto_offset_reset='earliest',
                         enable_auto_commit=False)

# 手動提交偏移量
for message in consumer:
    print(f"Received message: {message.value}")
    # 模擬處理訊息
    process_message(message)
    # 手動提交偏移量
    consumer.commit()

def process_message(message):
    # 實際處理訊息的邏輯
    pass

程式碼解密：

1. 建立 KafkaConsumer 例項：此段程式碼首先建立了一個 KafkaConsumer 例項，指定了要訂閱的主題、群組 ID、引導伺服器等引數。
2. 設定 auto_offset_reset：設定為 'earliest'，表示當沒有有效的偏移量時，從分割槽的最早訊息開始讀取。
3. 停用自動提交：設定 enable_auto_commit=False，表示將手動提交偏移量，以獲得更大的控制權。
4. 輪詢與處理訊息：透過迴圈遍歷 consumer 來接收訊息，並對每條訊息進行處理。
5. 手動提交偏移量：在成功處理每條訊息後，手動呼叫 consumer.commit() 提交偏移量，以確保不會遺漏任何訊息。
6. process_message 函式：此函式代表實際處理訊息的邏輯，需要根據具體需求實作。

Kafka系統可靠性驗證與最佳實踐

在建置Kafka系統時，為了確保系統的可靠性與資料完整性，我們需要經過多層面的驗證與測試。Kafka的設計允許高度的擴充套件性和容錯能力，但仍需要正確的組態和應用程式邏輯來滿足特定的可靠性需求。

維護狀態與處理重試錯誤

Kafka消費者需要正確處理重試錯誤和維護狀態。在傳統的pub/sub系統中，消費者會對每條訊息進行確認（“ack”）。然而，Kafka的消費者則是透過提交偏移量（offset）來標記已處理的訊息。如果在處理訊息時發生錯誤，消費者需要正確地處理這些錯誤，以避免丟失資料。

重試錯誤處理模式

1. 提交最後成功處理的偏移量：當遇到可重試的錯誤時，消費者可以提交最後成功處理的偏移量，並將未處理的訊息儲存在緩衝區中。然後，使用pause()方法暫停消費新的訊息，並不斷重試處理緩衝區中的訊息。

2. 寫入重試主題：另一種方法是將處理失敗的訊息寫入一個單獨的重試主題。然後，可以使用一個獨立的消費者群組來處理重試主題中的訊息，或者讓同一個消費者訂閱主主題和重試主題，並在重試之間暫停重試主題的消費。

維護跨多個輪詢呼叫的狀態

在某些應用程式中，需要跨多個poll()呼叫維護狀態。例如，計算移動平均值時，需要在每次輪詢Kafka取得新訊息後更新平均值。如果程式重新啟動，不僅需要從最後的偏移量開始消費，還需要還原相應的移動平均值。一個方法是將最新的累積值與偏移量一起寫入“結果”主題。這樣，當執行緒啟動時，可以取得最新的累積值並從上次中斷的地方繼續。

驗證系統可靠性

為了確保Kafka系統的可靠性，需要進行多層面的驗證：

1. 組態驗證：測試代理和客戶端的組態是否滿足可靠性需求。
2. 應用程式驗證：測試應用程式邏輯是否正確處理錯誤、提交偏移量和重新平衡事件。
3. 生產環境監控：在生產環境中監控應用程式，以確保其持續滿足可靠性需求。

組態驗證工具

Kafka提供了VerifiableProducer和VerifiableConsumer工具來幫助驗證組態。這些工具可以作為命令列工具執行，也可以嵌入自動化測試框架中。

• VerifiableProducer產生一系列包含數字的訊息，可以組態生產者的引數，如acks、retries和訊息生產速率。
• VerifiableConsumer消費事件並列印出消費的事件順序，以及提交和重新平衡的資訊。

測試場景

一些重要的測試場景包括：

• 長官者選舉：殺死長官者，觀察生產者和消費者還原正常需要多長時間。
• 控制器選舉：重新啟動控制器，觀察系統還原需要多長時間。
• 滾動重啟：逐一重啟代理，檢查是否丟失任何訊息。
• 不乾淨的長官者選舉測試：殺死所有副本，使其不同步，然後啟動一個不同步的代理，觀察系統還原需要做什麼。

應用程式驗證

在確認代理和客戶端組態滿足需求後，需要測試應用程式邏輯是否提供所需的保證。這包括測試自定義錯誤處理程式碼、偏移量提交和重新平衡監聽器等。

建議在開發過程中為應用程式編寫整合測試，並在各種故障條件下執行測試，例如：

• 客戶端失去與代理的連線
• 客戶端與代理之間的延遲很高
• 磁碟已滿
• 磁碟掛起（也稱為“棕色斷電”）
• 長官者選舉

透過這些驗證步驟，可以確保Kafka系統的可靠性和資料完整性，從而滿足特定的業務需求。

強化Kafka系統可靠性：測試、監控與最佳實踐

在構建可靠的Kafka系統時，測試和監控是兩個至關重要的環節。僅僅依靠Kafka本身的功能並不足以保證系統的可靠性，我們需要從應用架構、客戶端組態到主題和Broker組態進行全面考量。

滾動重啟與故障注入測試

為了確保Kafka叢集在各種異常情況下仍能保持穩定執行，我們需要進行滾動重啟測試和故障注入測試。

1. 滾動重啟測試：對Broker、消費者和生產者進行滾動重啟，以驗證系統在重啟過程中的穩定性和資料完整性。

   • 生產者和消費者的重啟可能會導致短暫的延遲或重複消費，但透過合理的組態（如delivery.timeout.ms和retries），可以最小化這些影響。

2. 故障注入測試：使用如Trogdor等工具模擬網路和磁碟故障，以測試系統的容錯能力。

   • 預先定義預期行為，並與實際測試結果進行對比，驗證系統是否符合設計預期。

生產環境監控

除了測試之外，持續監控生產環境中的Kafka叢集和客戶端也是保證系統可靠性的關鍵。

監控客戶端

1. 生產者監控：

   • 重點監控錯誤率和重試率，以識別潛在問題。
   • 檢查生產者日誌中WARN級別的錯誤訊息，如「Got error produce response…」，以判斷是否因錯誤導致訊息傳送失敗。

2. 消費者監控：

   • 重點關注消費者延遲（consumer lag），確保消費者不會持續落後於最新的訊息。
   • 使用Burrow等工具簡化消費者延遲的監控。

監控資料流

1. 確保資料及時消費：記錄生產者和消費者處理的事件數量，以及事件從生產到消費的時間延遲。
   • 結合事件時間戳，計算生產與消費之間的延遲。
   • 使用端對端監控系統（如Confluent Control Center）來跟蹤資料流的完整性。

監控Broker

1. 收集Broker指標：如FailedProduceRequestsPerSec和FailedFetchRequestsPerSec，以監控錯誤請求的變化趨勢。
   • 當Broker因維護而關閉並選舉新Leader時，預期會出現短暫的錯誤請求增加。持續上升的錯誤請求需要進一步調查。

詳細解說

• 應用架構設計：需考慮如何設計應用程式以利用Kafka實作可靠的訊息傳遞。
• 客戶端組態最佳化：根據實際需求調整生產者和消費者的組態，以提高系統的整體可靠性。
• 主題與Broker組態：合理組態主題的分割槽數、複製因子以及Broker的相關引數，以保證資料的安全性和可用性。

透過上述措施，可以顯著提升Kafka系統的可靠性和穩定性，為業務的持續執行提供堅實保障。