202535｜ Kafka架構與重要概念+冪等性+事務

好的！以下是關于 Kafka 架構 以及其 重要概念 的詳細介紹，結合 Mermaid 圖形 和表格，幫助你更好地理解各個概念的關系和作用。

Kafka 架構與重要概念

Kafka 是一個分布式消息系統，廣泛應用于日志收集、流處理、事件驅動架構等場景。它采用高吞吐量、可擴展的架構，支持多個組件進行數據的發布、訂閱、存儲和消費。

一、Kafka 架構圖（Mermaid 格式）

二、Kafka 重要概念詳細解釋

概念	說明
Broker	Kafka 服務器節點，負責存儲和管理消息。每個 Broker 處理多個 Topic 和 Partition 的數據讀寫。
ZooKeeper	分布式協調服務，Kafka 用它來管理集群的元數據，協調 Broker 狀態，選舉 Leader 等。
Producer	消息的生產者，將數據發布到指定的 Topic 中。
Consumer	消息的消費者，從 Kafka 中獲取數據并進行處理。
Consumer Group	由多個 Consumer 組成，Kafka 會確保每個分區在 Consumer Group 中只被一個消費者消費。
Topic	Kafka 中的主題，用于組織和分類消息。
Partition	每個 Topic 被劃分為多個分區，分區是 Kafka 存儲消息的基本單位。
Replica	分區的副本，確保數據的高可用性。每個分區可以有多個副本，其中一個為 Leader。
Offset	Kafka 中每條消息在分區內的唯一標識，消費者通過 Offset 來追蹤自己消費的進度。

三、Kafka 架構中的各個組件

1. Broker（Kafka 代理）

定義：Kafka 中的 Broker 是負責存儲消息并處理客戶端請求的節點。每個 Broker 負責多個 Topic 和 Partition 的數據讀寫。
角色：Kafka 集群由多個 Broker 組成，集群中的 Broker 節點共同工作，確保數據的存儲、消費和高可用性。
作用：Broker 接受來自 Producer 的消息，將其存儲在本地磁盤中，并提供 Consumer 按需讀取。

2. ZooKeeper（協調服務）

定義：ZooKeeper 是一個分布式協調服務，用于管理 Kafka 集群的元數據和協調 Broker 的狀態。Kafka 使用 ZooKeeper 來進行 Broker 的注冊、Leader 選舉等管理操作。
作用：它幫助 Kafka 保持一致性、選舉 Leader 和管理集群狀態。ZooKeeper 集群中可以有多個節點，提供高可用性。

3. Producer（生產者）

定義：Producer 是向 Kafka 集群發送消息的客戶端。它將消息發送到指定的 Topic 中。
作用：Producer 可以選擇發送到特定的 Partition，也可以選擇由 Kafka 自動分配 Partition。
分區選擇：Producer 可以通過指定一個 key 來確保某一類消息始終發送到同一個分區，或者由 Kafka 根據負載均衡策略自動選擇。

4. Consumer（消費者）

定義：Consumer 是從 Kafka 中讀取消息并處理的客戶端。它可以訂閱一個或多個 Topic，讀取分區中的消息。
作用：Consumer 從 Kafka 中消費消息，通常會從消息的 Offset 開始讀取。消費者通常會在自己的 Offset 位置繼續消費。

5. Consumer Group（消費組）

定義：Consumer Group 是由多個 Consumer 組成的一組消費者。每個 Consumer Group 內的消費者共同消費同一個 Topic 的消息，但每個分區的消息只有一個消費者消費。
作用：通過消費組，Kafka 可以實現消息的并行消費。每個消費組都會維護自己的 Offset 獨立于其他消費組。

6. Topic（主題）

定義：Topic 是 Kafka 中的消息分類標簽。Producer 將消息發送到特定的 Topic，Consumer 可以訂閱一個或多個 Topic 來消費消息。
作用：Topic 是 Kafka 消息的邏輯分組，可以幫助管理不同種類的消息流。

7. Partition（分區）

定義：每個 Topic 會被劃分成多個 Partition，Partition 是 Kafka 存儲消息的最小單位。每個分區存儲消息的順序，并且每個 Partition 都有一個 Leader 和若干個 Follower 副本。
作用：分區提供了水平擴展的能力，可以增加分區來處理更高的消息吞吐量和并發消費。

8. Replica（副本）

定義：每個分區都有多個副本（Replica），其中一個副本是 Leader，負責處理所有的讀寫請求。其他副本是 Follower，負責同步數據。
作用：副本保證了 Kafka 集群的數據高可用性和容錯性。即使某個 Broker 節點失敗，其他副本仍然可以保證數據不丟失。

9. Offset（偏移量）

定義：Offset 是 Kafka 中每條消息在分區內的唯一標識。每個消費者在消費消息時都會記錄自己的 Offset 位置。
作用：Offset 使得 Kafka 的消息消費具有高可控性。消費者可以從特定的 Offset 開始消費消息，也可以重新從頭消費歷史消息。

Kafka生產者冪等性

Kafka 生產者的冪等性功能是為了保證消息的 一次性寫入。即使在網絡超時、生產者重試等情況下，消息依然能夠保證 只被寫入一次。為了實現這一目標，Kafka 利用一系列機制來控制消息的重復發送，并對每條消息進行唯一標識。

Kafka 生產者冪等性核心概念

Producer ID (PID): 每個 Kafka 生產者實例都會在啟動時獲取一個唯一的 Producer ID。這是 Kafka 用來區分不同生產者的標識符，確保每個生產者發送的消息在 Kafka 集群中都有唯一標識。
Sequence Number (序列號): 每個生產者發送的消息都有一個遞增的序列號，Sequence Number 是 Kafka 用來判定消息是否重復的關鍵參數。序列號的遞增可以確保即使是同一個生產者發送多條消息，Kafka 也能區分它們。
Message Key & Partition: 消息的 Key 用來決定消息將發送到哪個分區，確保同一個 Key 的消息總是發送到同一個分區。分區內的消息順序對于冪等性沒有影響，但消息的重復性檢查依賴于 Producer ID 和 Sequence Number。

生產者冪等性工作流程

為了解釋 Kafka 是如何確保消息不被重復寫入，我們可以將其工作過程分解為幾個步驟：

生產者初始化：
- 每個生產者在啟動時會向 Kafka 集群申請一個唯一的 Producer ID (PID)，并且會從 Kafka 中獲得一個初始的 Sequence Number。
- 生產者為每條發送的消息附上遞增的序列號。
消息發送：
- 生產者向 Kafka 發送消息時，消息攜帶兩個關鍵字段：Producer ID 和 Sequence Number。這些字段幫助 Kafka 區分不同生產者發送的消息，以及同一生產者發送的不同消息。
消息接收與存儲：
- Kafka Broker 在接收到消息后，會先檢查 Producer ID 和 Sequence Number。如果該消息的 Sequence Number 對應的消息已經存在，說明這是一個 重復消息，則 Kafka 會忽略該消息，不進行存儲。
- 如果消息是新的，Kafka 會將其存儲，并將該消息的 Producer ID 和 Sequence Number 存儲在元數據中。
消息重試：
- 在網絡延遲或其他故障的情況下，生產者可能會重試發送相同的消息。由于生產者會附帶相同的 Producer ID 和 Sequence Number，Kafka Broker 會識別到這是同一條消息，并忽略重試請求，確保消息只被寫入一次。
消息確認：
- 在消息成功寫入 Kafka 后，Broker 會將寫入成功的確認返回給生產者。若發生故障或超時，生產者會繼續重試，直到成功或者達到重試次數的上限。

Kafka 生產者冪等性詳細流程圖

為了讓你更直觀地理解這一過程，下面是一個更詳細的圖解，展示了生產者發送消息、重試以及如何避免重復消息寫入的流程：

工作原理解讀

1. 生產者啟動時

當生產者啟動時，Kafka 為該生產者分配一個 Producer ID（如 123），并且給該生產者分配一個遞增的 Sequence Number（從 0 開始）。每個消息都攜帶這些信息，確保 Kafka 能識別該生產者的每一條消息。

2. 生產者發送消息（消息 1）

生產者發送 消息 1 到 Kafka 集群，消息的內容包括：Producer ID: 123 和 Sequence Number: 0。Kafka 會根據這個信息將消息存儲并傳遞給消費者。

3. 網絡故障與重試

由于網絡延遲或其他原因，生產者可能會重試發送相同的消息。此時，生產者再次發送 消息 1（Producer ID: 123, Sequence: 0）。Kafka 在收到這條消息時，會檢查消息的 Producer ID 和 Sequence Number，發現這條消息已經存在，因此會忽略重復的消息。

4. 發送新的消息（消息 2）

當生產者發送 消息 2 時，Sequence Number 增加為 1，Kafka 會識別為新消息，并將其存儲和傳遞給消費者。

Kafka 冪等性實現的關鍵因素

Producer ID (PID)：
- 每個生產者在 Kafka 中都有一個唯一的 Producer ID，它用來標識消息的來源。即使網絡重試，Kafka 也能根據 Producer ID 來區分不同的生產者。
Sequence Number：
- 每個生產者發送的消息都有一個遞增的序列號。這個序列號使 Kafka 能夠區分同一生產者發送的不同消息，確保消息的順序性。
去重機制：
- Kafka 使用 Producer ID 和 Sequence Number 來判斷消息是否重復。如果收到相同 Producer ID 和 Sequence Number 的消息，Kafka 會認為這是一條重復消息，進而忽略它。
跨分區的一致性：
- Kafka 的冪等性機制是 跨分區的，即使消息被發送到多個分區，只要生產者的 Producer ID 和 Sequence Number 相同，Kafka 就會確保消息不重復寫入。

生產者冪等性機制的優缺點

優點

保證消息準確性：即使發生網絡重試、故障等問題，生產者也能保證消息 只寫一次，避免了數據的重復和不一致。
提升容錯能力：在發生故障或超時的情況下，生產者能夠安全重試，并且 Kafka 會確保不會重復寫入消息，增強了系統的容錯性。
避免消費者重復消費：冪等性機制確保了消費者不會收到重復的消息，避免了重復消費和數據錯誤。

缺點

性能開銷：啟用冪等性會增加一定的性能開銷，尤其在高吞吐量的場景下，Kafka 需要進行更多的檢查和存儲操作來確保消息唯一性。
對多生產者的支持有限：冪等性機制僅對單一生產者實例有效，不同生產者間無法共享冪等性狀態。

Kafka 生產者冪等性配置示例

以下是一個典型的 Kafka 生產者配置，啟用了冪等性功能：

java復制編輯
Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 啟用冪等性
properties.put("acks", "all");  // 等待所有副本確認
properties.put("retries", Integer.MAX_VALUE);  // 設置最大重試次數為最大值
properties.put("max.in.flight.requests.per.connection", 1);  // 防止并發請求導致亂序
properties.put("enable.idempotence", "true");  // 啟用冪等性KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

配置詳解：

acks=all：等待所有副本確認后才返回確認，提供了最佳的消息可靠性。
retries=Integer.MAX_VALUE：允許生產者無限次重試，直到成功。通過冪等性機制，Kafka 確保即使消息重試，也不會重復寫入。
max.in.flight.requests.per.connection=1：為了確保消息的順序性，生產者一次只允許發送一個未確認的消息。
enable.idempotence=true：啟用冪等性，確保消息只寫一次，即使發生重試，消息也不會重復。

KafKa事務

Kafka 事務簡介

Kafka 的事務是一個強大的特性，它允許用戶將多個消息的發送操作封裝在一個 原子操作 中。通過事務，生產者可以確保多條消息的發送要么全部成功，要么全部失敗。Kafka 的事務能夠保證數據一致性，特別是在分布式環境下，避免了部分消息寫入而其他消息丟失的問題。

Kafka 事務不僅僅用于生產者，也為消費者提供了 精確一次消費（Exactly Once Semantics, EOS）的能力。事務在 Kafka 中的應用解決了消息重復消費和漏消費的難題，提升了消息流的可靠性。

Kafka 事務的核心特點

原子性（Atomicity）： Kafka 事務保證一組消息的發送要么全部成功，要么全部失敗。若消息在事務內的某些操作失敗，那么整個事務會被回滾，確保消息的原子性。
可靠性（Durability）：一旦事務提交，Kafka 會保證消息持久化，并且可以恢復事務內的消息。
一致性（Consistency）： Kafka 確保事務內的消息要么全都提交，要么全都回滾，避免了數據不一致的情況。
隔離性（Isolation）： Kafka 確保事務內的消息在事務提交之前對消費者是不可見的。即消費者只能看到已經提交的消息，未提交的事務消息對消費者是不可見的。

Kafka 事務的實現

Kafka 使用以下兩種類型的標記來管理事務：

事務開始：生產者發出開始事務的信號，所有后續發送的消息都屬于這個事務。
事務提交：生產者發出提交事務的信號，Kafka 將事務中的所有消息持久化，確保消息可供消費者讀取。
事務中止（回滾）：如果生產者遇到錯誤并決定放棄事務，可以發出中止事務的信號，所有事務內的消息都不會被提交。

Kafka 會在事務日志中記錄每個事務的狀態。事務狀態包括提交、回滾或未決（進行中）。

Kafka 事務的關鍵配置

acks：
- 設置為 all，確保所有副本都確認消息，以提高消息可靠性和容錯性。
transactional.id：
- 這是 Kafka 生產者的一個關鍵配置，標識一個生產者實例的事務標識符。每個事務性生產者都需要設置這個唯一的標識符，以便 Kafka 能追蹤事務狀態。
retries：
- 在事務中，重試機制非常重要，設置 retries 可以確保生產者在事務中的消息如果失敗，可以重試。
transaction.timeout.ms：
- 這個配置用于設置事務的超時時間，默認值是 60000 毫秒。若一個事務未在指定時間內提交或回滾，Kafka 會認為事務超時并自動中止事務。
enable.idempotence：
- 在啟用事務的同時，還需要啟用冪等性，這樣可以確保即使在網絡中斷或生產者重試的情況下，也不會重復發送相同的消息。

Kafka 事務的生產者使用流程

開啟事務：在生產者代碼中，需要使用 beginTransaction() 方法顯式地開始一個事務。
發送消息：在事務中發送消息時，所有發送的消息都會被標記為事務的一部分，直到事務被提交或回滾。
提交事務：通過 commitTransaction() 方法提交事務，Kafka 會將事務中的所有消息持久化。
回滾事務：如果在事務過程中發生了錯誤，生產者可以使用 abortTransaction() 方法回滾事務，所有未提交的消息都不會被持久化。

事務的生產者代碼示例

以下是一個使用 Kafka 事務的 Java 代碼示例，演示如何開啟事務、發送消息并提交事務。

Properties props = new Properties();
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");
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);
props.put("transactional.id", "my-transaction-id");// 創建 KafkaProducer 實例
KafkaProducer<String, String> producer = new KafkaProducer<>(props);try {// 開始事務producer.beginTransaction();// 發送消息producer.send(new ProducerRecord<>("my-topic", "key1", "value1"));producer.send(new ProducerRecord<>("my-topic", "key2", "value2"));// 提交事務producer.commitTransaction();
} catch (ProducerFencedException | OutOfMemoryError | KafkaException e) {// 在發生異常時回滾事務producer.abortTransaction();
} finally {producer.close();
}

代碼解釋：

初始化生產者：配置 bootstrap.servers、acks 等基本參數，并設置 transactional.id 來啟用事務。
開始事務：通過 beginTransaction() 方法啟動一個事務。
發送消息：消息會被添加到當前事務中，直到事務提交或回滾。
提交事務：commitTransaction() 方法將事務中的所有消息持久化到 Kafka 集群中。
異常處理和回滾：如果發生異常（如網絡中斷、寫入失敗等），調用 abortTransaction() 方法回滾事務，確保事務內的消息不被寫入。

Kafka 事務的消費者

Kafka 的事務不僅對生產者有效，還影響消費者的行為。啟用事務后，消費者只能看到已經提交的消息，這意味著未提交的事務對消費者是不可見的。這保證了 精確一次消費（Exactly Once Semantics，EOS），確保消費者不會消費到重復或不完整的數據。

消費者的事務性讀取：

如果消費者消費到一個尚未提交的消息，它會等待直到該消息所在的事務提交。
如果事務被回滾，消費者不會讀取到該事務中的消息。

啟用精確一次消費：

為了啟用精確一次的消費，消費者需要設置 isolation.level 配置項，通常設置為 read_committed，這意味著消費者只會讀取已提交的消息。

Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "my-consumer-group");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("isolation.level", "read_committed");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("my-topic"));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.println("Consumed: " + record.value());}
}