kafka源碼分析
『壹』 源碼解析kafka刪除topic
源碼解析Kafka刪除Topic的詳細流程
本文將詳細介紹在Kafka 0.8.2.2版本中刪除Topic的過程，分為幾個關鍵步驟。
刪除步驟
1. 配置參數
要執行Topic刪除，首先需要在Broker上配置delete.topic.enable為true，這是刪除操作的必要條件。
2. 執行刪除操作
在命令行中使用bin/kafka-topics.sh --zookeeper zk_host:port/chroot --delete --topic my_topic_name。若不設置該參數，Topic只是被標記為刪除，數據不會真正清除，重啟Kafka是清除數據的最佳策略。
涉及的關鍵類
- PartitionStateMachine 管理分區狀態，包括NonExistentPartition、NewPartition、OnlinePartition和OfflinePartition等。
- ReplicaManager 負責管理副本操作，如讀寫和刪除。
- ReplicaStateMachine 副本狀態機，控制副本的五種狀態變化，如NewReplica、OnlineReplica等。
- TopicDeletionManager 負責Topic刪除的全局管理。
源碼分析
刪除流程分為四部分：
- 客戶端命令影響
- 不配置delete.topic.enable的流程
- 配置delete.topic.enable的流程
- 手動刪除
1. 客戶端命令觸發刪除操作，創建Zookeeper節點。
2. 不配置delete.topic.enable時，Topic刪除操作受監聽器控制，僅標記刪除，不執行數據清除。
3. 配置delete.topic.enable後，會檢查某些條件（如副本故障、分區重分配等），然後更新刪除任務隊列並啟動刪除線程，徹底清除數據。
4. 手動刪除Zookeeper節點和磁碟數據，但可能導致內存數據未清除，是否安全需進一步測試。
總結：
Kafka Topic的刪除是基於Zookeeper的事件驅動模型。刪除命令觸發刪除操作，涉及數據的清除包括解除監聽、清理內存、磁碟副本的刪除以及Zookeeper節點的移除。配置delete.topic.enable為true是確保數據完全清除的關鍵步驟。
『貳』 源碼解析kafka刪除topic
本文將詳細解析Kafka中刪除topic的具體實現與過程。以Kafka 0.8.2.2版本為例，主要涉及以下幾個關鍵點：
### 1. 刪除topic的步驟
刪除一個topic需要完成以下兩個關鍵步驟：
配置刪除參數：確保`delete.topic.enable`參數在Broker端設置為`True`。
執行刪除命令：使用`bin/kafka-topics.sh`命令，指定`--delete`選項和目標topic名稱。
若不配置`delete.topic.enable`為`true`，雖然topic會被標記為刪除狀態，但其在Kafka Broker內存中的數據並未被清除。此時，最理想的策略是配置`delete.topic.enable`為`true`，然後重啟Kafka服務。
### 2. 重要類介紹
涉及刪除topic的幾個關鍵類包括：
**PartitionStateMachine**：代表分區的狀態機，決定分區的當前狀態和狀態轉移。其包含四種狀態：NonExistentPartition、NewPartition、OnlinePartition和OfflinePartition。
**ReplicaManager**：負責管理當前機器的所有副本，處理讀寫、刪除等具體動作。讀寫過程中，先獲取partition對象，再獲取Replica對象，然後獲取Log對象，通過其管理的Segment對象進行數據的寫入和讀取。
**ReplicaStateMachine**：副本的狀態機，決定副本的當前狀態和狀態之間的轉移。一個副本可以處於以下狀態：NewReplica、OnlineReplica、OfflineReplica、ReplicaDeletionStarted、ReplicaDeletionSuccessful和ReplicaDeletionIneligible。
**TopicDeletionManager**：管理topic刪除狀態機，其通過`/admin/delete_topics`目錄下的節點來執行刪除命令。
### 3. 源碼解析刪除過程
刪除topic的過程可以分為以下幾個階段：
客戶端執行刪除命令時，在`kafka-topics.sh`中調用`kafka-run-class.sh kafka.admin.TopicCommand`執行刪除操作。
在不配置`delete.topic.enable`的情況下，整個刪除流程會檢查`delete.topic.enable`是否為`true`。若為`false`，則不會執行刪除操作。若為`true`，則繼續執行。
配置`delete.topic.enable`為`true`時，會停止刪除topic，並處理三種特殊情況：副本下線、分區重新分配中和首選副本選舉中。`enqueueTopicsForDeletion`方法更新需要刪除的topic集合並激活刪除線程。
刪除線程`DeleteTopicsThread`的`doWork`方法中，`completeDeleteTopic`方法負責解除分區變更監聽、刪除Zookeeper中的具體節點信息、清理磁碟數據並更新內存數據結構，如從副本狀態機中移除分區信息。
在首次清除過程中，刪除線程`DeleteTopicsThread`的`doWork`方法中，Kafka執行了完整的日誌刪除流程。
### 4. 手動刪除Zookeeper信息和磁碟數據
雖然手動刪除Zookeeper的topic信息和磁碟數據可以實現刪除操作，但可能會留下內存中的部分數據未清除。具體實施時需謹慎，以避免潛在的問題。
### 5. 總結
Kafka中topic的刪除流程主要基於Zookeeper實現，通過創建`/admin/delete_topics`目錄下的節點來觸發刪除命令。配置`delete.topic.enable`為`true`，並重啟Kafka服務，可以確保topic信息被徹底刪除。手動刪除Zookeeper信息和磁碟數據雖可實現刪除，但可能導致內存中的部分數據未清除，建議在實施前進行充分測試。
『叄』 一文解密Kafka，Kafka源碼設計與實現原理剖析，真正的通俗易懂
Apache Kafka （簡稱Kafka ）最早是由Linkedln開源出來的分布式消息系統，現在是Apache旗下的一個子項目，並且已經成為開冊、領域應用最廣泛的消息系統之 Kafka社區也非常活躍，從 版本開始， Kafka 的標語已經從「一個高吞吐量、分布式的消息系統」改為「一個分布式的流平台」
關於Kafka，我打算從入門開始講起，一直到它的底層實現邏輯個原理以及源碼，建議大家花點耐心，從頭開始看，相信會對你有所收獲。
作為 個流式數據平台，最重要的是要具備下面 個特點
消息系統：
消息系統 也叫作消息隊列）主要有兩種消息模型：隊列和發布訂Kafka使用消費組（ consumer group ）統 上面兩種消息模型 Kafka使用隊列模型時，它可以將處理 作為平均分配給消費組中的消費者成員
下面我們會從 個角度分析Kafka 的幾個基本概念，並嘗試解決下面 個問題
消息由生產者發布到 fk 集群後，會被消費者消費 消息的消費模型有兩種：推送模型（ pu和拉取模型（ pull 基於推送模型的消息系統，由消息代理記錄消費者的消費狀態 消息代理在將消息推送到消費者後 標記這條消息為已消費
但這種方式無法很好地保證消息的處理語義 比如，消息代理把消息發送出去後，當消費進程掛掉或者由於網路原因沒有收到這條消息時，就有可能造成消息丟失（因為消息代理已經 這條消息標記為自己消費了，但實際上這條消息並沒有被實際處理） 如果要保證消息的處理語義，消息代理發送完消息後，要設置狀態為「已發送」，只有收到消費者的確認請求後才更新為「已消費」，這就需要在消息代理中記錄所有消息的消費狀態，這種做法也是不可取的
Kafka每個主題的多個分區日誌分布式地存儲在Kafka集群上，同時為了故障容錯，每個分區都會以副本的方式復制到多個消息代理節點上 其中一個節點會作為主副本（ Leader ），其 節點作為備份副本（ Follower ，也叫作從副本）
主副本會負責所有的客戶端讀寫操作，備份副本僅僅從主副本同步數據 當主副本 IH 現在故障時，備份副本中的 副本會被選擇為新的主副本 因為每個分區的副本中只有主副本接受讀寫，所以每個服務端都會作為某些分區的主副本，以及另外一些分區的備份副本這樣Kafka集群的所有服務端整體上對客戶端是負載均衡的
消息系統通常由生產者「pro ucer 消費者（ co sumer ）和消息代理（ broke 大部分組成，生產者會將消息寫入消息代理，消費者會從消息代理中讀取消息 對於消息代理而言，生產者和消費者都屬於客戶端：生產者和消費者會發送客戶端請求給服務端，服務端的處理分別是存儲消息和獲取消息，最後服務端返回響應結果給客戶端
新的生產者應用程序使用 af aP oce 對象代表 個生產者客戶端進程 生產者要發送消息，並不是直接發送給 務端 ，而是先在客戶端 消息放入隊列 然後 一個 息發送線程從隊列中消息，以 鹽的方式發送消息給服務端 Kafka的記 集器（ Reco dACCUl'lUlato ）負責緩存生產者客戶端產生的消息，發送線程（ Sende ）負責讀取 集器的批 過網路發送給服務端為了保證客戶端 絡請求 快速 應， Kafka 用選擇器（ Selecto 絡連接 讀寫 理，使網路連接（ Netwo kCl i.ent ）處理客戶端 絡請求
追加消息到記錄收集器時按照分區進行分組，並放到batches集合中，每個分區的隊列都保存了將發送到這個分區對應節點上的 記錄，客戶端的發送線程可 只使用 Sende 線程迭 batches的每個分區，獲取分區對應的主劇本節點，取出分區對應的 列中的批記錄就可以發送消息了
消息發送線程有兩種消息發送方式 按照分區直接發送 按照分區的目標節點發迭 假設有兩台伺服器， 題有 個分區，那麼每台伺服器就有 個分區 ，消息發送線程迭代batches的每個分 接往分區的主副本節點發送消息，總共會有 個請求 所示，我 先按照分區的主副本節點進行分組， 屬於同 個節點的所有分區放在一起，總共只有兩個請求做法可以大大減少網路的開銷
消息系統由生產者 存儲系統和消費者組成 章分析了生產者發送消息給服務端的過程，本章分析消費者從服務端存儲系統讀取生產者寫入消息的過程 首先我 來了解消費者的 些基礎知識
作為分布式的消息系統， Kafka支持多個生產者和多個消費者，生產者可以將消息發布到集群中不同節點的不同分區上；「肖費者也可以消費集群中多個節點的多個分區上的消息 寫消息時，多個生產者可以 到同 個分區 讀消息時，如果多個消費者同時讀取 個分區，為了保證將日誌文件的不同數據分配給不同的消費者，需要採用加鎖 同步等方式，在分區級別的日誌文件上做些控制
相反，如果約定「同 個分區只可被 個消費者處理」，就不需要加鎖同步了，從而可提升消費者的處理能力 而且這也並不違反消息的處理語義：原先需要多個消費者處理，現在交給一個消費者處理也是可以的 3- 給出了 種最簡單的消息系統部署模式，生產者的數據源多種多樣，它們都統寫人Kafka集群 處理消息時有多個消費者分擔任務 ，這些消費者的處理邏輯都相同， 每個消費者處理的分區都不會重復
因為分區要被重新分配，分區的所有者都會發生變 ，所以在還沒有重新分配分區之前 所有消費者都要停止已有的拉取錢程 同時，分區分配給消費者都會在ZK中記錄所有者信息，所以也要先刪ZK上的節點數據 只有和分區相關的 所有者 拉取線程都釋放了，才可以開始分配分區
如果說在重新分配分區前沒有釋放這些信息，再平衡後就可能造成同 個分區被多個消費者所有的情況 比如分區Pl 原先歸消費者 所有，如果沒有釋放拉取錢程和ZK節點，再平衡後分區Pl 被分配給消費者 了，這樣消費者 和消費者 就共享了分區Pl ，而這顯然不符合 fka 中關於「一個分區只能被分配給 個消費者」的限制條件 執行再平衡操作的步驟如下
如果是協調者節點發生故障，服務端會有自己的故障容錯機制，選出管理消費組所有消費者的新協調者節，點消費者客戶端沒有權利做這個工作，它能做的只是等待一段時間，查詢服務端是否已經選出了新的協調節點如果消費者查到現在已經有管理協調者的協調節點，就會連接這個新協調節，哉由於這個協調節點是服務端新選出來的，所以每個消費者都應該重新連接協調節點
消費者重新加入消費組，在分配到分區的前後，都會對消費者的拉取工作產生影響 消費者發送「加入組請求」之前要停止拉取消息，在收到「加入組響應」中的分區之後要重新開始拉取消息時，為了能夠讓客戶端應用程序感知消費者管理的分區發生變化，在加入組前後，客戶端還可以設置自定義的「消費者再平衡監聽器」，以便對分區的變化做出合適的處理