前言

本次復習主要是看的胡夕老師的Kafka核心技術與實戰，講得很好，主要是查缺補漏。

Kafka是一種消息引擎系統。負責的事情也很簡單：

1. 系統A將消息發送給Kafka。
2. 系統B從Kafka中讀取系統A發送的消息。

Kafa對于傳輸的消息編碼格式的選擇：純二進制的字節序列。

Kafka中傳輸消息的模型有兩種：

• 點對點模型（消息隊列模型）：系統A發送的消息只能被系統B接收。
• 發布 / 訂閱模型：擁有主題Topic概念，可以有多個發布者Publisher向相同的主題發送消息。可以同時被多個訂閱者Subscriber接收。

為什么要使用Kafka？（或者A系統為什么不能直接把消息發送給B系統？）回答：削峰填谷

所謂的“削峰填谷”就是指緩沖上下游瞬時突發流量，使其更平滑。對于那種發送能力很強的上游系統，如果沒有消息引擎的保護，“脆弱”的下游系統可能會直接被壓垮導致全鏈路服務“雪崩”。

Kafka和其他MQ的區別在哪？比如RabbitMQ和Kafka如何選擇？

1. RabbitMQ屬于比較傳統的消息隊列系統，支持標準的消息隊列協議：AMQP、STOMP、MQTT等。
2. 因此倘若具體的應用需要支持這些協議，那么選擇RabbitMQ。
3. RabbitMQ還支持比較復雜的消費路由，而Kafka不支持。

---

備注：消息傳輸協議和Http這一類網絡通信協議是不同的！一般兩個進程之間進行數據流的交互有三種方式：

• 通過數據庫的讀寫：進程1寫數據到數據庫，進程2從數據庫中讀數據。
• 通過服務調用：RPC調用或者Rest調用。HTTP協議通常就作為rest方式的底層通訊協議。
• 通過消息傳遞的方式：進程1發送消息給中間件，例如Kafka。進程2從中間件中讀取消息。而消息傳輸協議就運用與此。

---

那竟然RPC調用和發MQ，其本質都是不同進程之間數據流的一個交互，那么MQ和RPC的區別又是什么？

1. MQ有自己的buffer，能夠對抗過載（overloaded）。
2. MQ支持重試retry。
3. MQ允許發布/訂閱功能。

一. Kafka 相關術語

Kafka屬于分布式的消息引擎系統。主要提供一套完備的消息發布和訂閱解決方案。

首先來說下Kafka的幾個最最基礎的概念：

• 消息（Record）：Kafka處理的主要對象。
• 主題（Topic）：承載消息的邏輯容器，一般用來區分具體的業務。
• 生產者（Producer）：向主題發布信息的應用程序。可以向一個或者多個主題發送消息。
• 消費者（Consumer）：從主題訂閱信息的應用程序。可以同時訂閱多個主題的消息。
• 消費者組（Consumer Group）：為了提高吞吐量，將多個消費者實例共同組成一個組，消費同一個主題。但是主題中的每個分區只會被組內的一個消費者實例消費。

其中，生產者和消費者這樣的應用程序，我們統一稱之為客戶端Clients。那么隨之對應的，也有服務端這么一個概念。

1.1 實現高可用的手段

Kafka的服務器端由稱之為Broker的進程構成。Broker有什么作用呢？

1. Broker負責接收和處理客戶端發送過來的請求。
2. 負責對消息進行持久化。

高可用手段一：一般Kafka將不同的Broker分散運行在不同的機器上。這樣哪怕某臺機器宕機，那么其他的機器的Broker也能夠對外提供服務。

---

高可用手段二：備份機制（Replication）。

備份機制的本質就是對數據進行拷貝。Kafka中，對這些數據拷貝稱之為副本（replica），副本分為兩種：

• 領導者副本：Leader Replica，負責對外提供服務，與客戶端進行交互。
• 追隨者副本：Follower Replica，只負責更新領導者副本中的數據。

生產者總是向領導者副本寫消息，而消費者也總是從領導者副本中讀取消息。追隨者副本，注意：不提供對外服務，不提供對外服務，不提供對外服務，重要的事情說三遍。只負責與領導者副本保持同步。

緊接著，為了避免單臺Broker上無法容納更多的數據。Kafka對副本進行分割，也就是所謂的分區（Partitioning）。

Kafka中將每個主題劃分為多個分區。每個分區包含一組有序的消息日志。生產者生產的每條消息只會被發送到一個分區中。

1.2 Kafka的三層消息架構

有了主題、副本、分區概念之后，Kafka可以分為三層的消息架構：

• 主題層：每個主題分為M個分區，每個分區又可以分成N個副本。
• 分區層：每個分區的N個副本中，包含1個領導者副本，負責對外提供服務。N-1個追隨者副本，負責提供冗余數據。
• 消息層：分區中包含若干條消息，每條消息的位移從0開始并依次遞增。

1.3 Kafka如何持久化數據

Kafka使用消息日志來保存數據：一種在磁盤上只能通過追加寫（Append-only） 的方式來記錄消息的物理文件。

追加寫的好處：

1. 將緩慢的隨機IO操作，改為性能較好地順序IO寫操作。

同時，Kafka通過日志段（Log Segment）機制定期的刪除消息來回收磁盤空間。

• Kafka底層，一個日志會進一步細分為多個日志段。
• 消息在追加寫的時候會記錄到當前最新的日志段中。
• 當寫滿一個日志段后，Kafka就會自動分配一個新的日志段，并將老的日志段封存。
• 通過定時任務來檢查老的日志段是否能夠被刪除，達到磁盤空間回收的目的。

其他的概念：

• 重平衡（Rebalance）：消費者組內某個消費者實例掛掉后，將分區的所有權從該消費者轉移到另一個消費者。
• 消費者位移（Consumer Offset）：每個消費者在消費消息的過程中必然需要有個字段記錄它當前消費到了分區的哪個位置上。也就是所謂的消費者位移。

1.4 常見問題

為什么Kafka不支持主從分離？

1. 我們知道Redis和Mysql都支持主從的一個讀寫分離。主從分離只是一種架構設計，往往適用于那種讀多寫少的應用場景。 而Kafka的主要應用場景是消息引擎，用于頻繁地生產/消費消息。不屬于典型的讀多寫少場景，因此讀寫分離方案在這個場景下就不太合適。
2. Kafka的副本機制方面，使用的是異步消息的拉取。存在leader和follower之間的不一致性。 若采用讀寫分離。必然需要去處理數據的不一致性問題。例如：如何實現read-your-writes、如何保證單調讀，如何處理消息因果序顛倒的問題等。
3. 此外，Kafka的一個數據性質和Mysql不一樣，Kafka中的數據存在著消費的概念，可以看做流數據，當然并不是說消費了這個數據就不在了，對應的Kafka中有著消費者位移的概念，可以看做消費的進度在哪了。而數據庫就不存在這樣的概念。
4. Kafka中的分區可以分散到各個Broker上，也能做到負載均衡的效果。也沒必要通過主從讀寫分離的方式來負載均衡。

---

read-your-writes是什么？

1. 當生產者成功向Kafka中寫入消息的時候。馬上使用消費者去讀取剛剛生產的消息。
2. 那么此刻我希望能讀到我剛剛寫的最新的數據。
3. 這里再說一下關于主從問題，倘若Kafka允許從副本提供對外服務，那么由于主從之間數據更新的異步性，無法保證客戶端從追隨者副本中讀取到的信息是最新的。也就無法保證read-your-writes。

---

單調讀的概念？

• 對于一個消費者而言，在多次消費的時候，不會存在某一個消息時而存在時而不存在的情況。

二. Kafka集群部署方案

首先從操作系統的角度來看：目前比較常見的操作系統有三種：Linux、Windows 和 macOS。

2.1 操作系統的選擇

考慮操作系統和Kafka之間的適配性，Linux是更好的選擇，主要從以下三點來說：

• IO模型的使用（復習Kafka的時候，特地去復習了下IO模型：IO知識復習）。
• 數據網絡傳輸效率。
• 社區支持度。

首先對于IO模型的使用：

1. Kafka的客戶端底層使用了Java的selector。
2. selector在Linux上的實現機制就是epoll，而在Windows平臺上的實現機制是select。
3. 因此在這一點上，Kafka部署在Linux上是有優勢的，能夠獲得更高效的IO性能。epoll優勢更大。

其次對于網絡傳輸效率方面的考慮：若Kafka在Linux上部署，能夠享受到零拷貝技術帶來的快速數據傳輸特性。

關于零拷貝的復習：Linux - 零拷貝技術

最后再是社區的支持度：一般人都會選擇Linux來搭建。 社區目前對 Windows 平臺上發現的 Kafka Bug 不做任何承諾。

2.2 磁盤的選擇

我們都知道，固態硬盤比普通的磁盤要快，這里至于為什么快，我感覺也沒必要去深入了解，知道就好。而Kafka集群所需要的存儲空間還是比較大的，需要考慮這么幾個因素：

1. 新增的消息數。
2. 消息留存時間。
3. 平均消息的大小。
4. 副本的數量。
5. 消息是否啟用壓縮功能。

而固態硬盤的優勢大，但是成本要高，機械硬盤的成本低，但是容易損壞。這里給個參考。

有條件的情況下，無腦選固態硬盤。
若條件比較苛刻，選擇機械硬盤的問題不大。理由如下：

1. Kafka雖然大量使用磁盤空間，但是使用的方式大部分是順序讀寫IO。因此一定程度上規避了機械磁盤的劣勢（隨機IO讀寫慢）。
2. Kafka有自己的冗余機制（副本）來提高可靠性，一定程度上彌補了硬盤容易損壞的缺點。

2.3 重要集群參數

2.3.1 Broker端參數

針對日志存儲的相關參數：

• log.dirs：沒有默認值（因此必須手動指定），用于指定Broker需要使用的若干個文件目錄路徑，每個路徑之間用 逗號 分割。 最好將目錄掛載到不同的物理磁盤上，這樣就有兩個好處：

1. 提升讀寫性能：多塊物理磁盤同時讀寫比單塊磁盤要有更高的吞吐量。
2. 實現故障轉移。

• log.dir：表示單個路徑，用于補充上一個參數用的。

---

Kafka的運行一般還依賴于zookeeper，zookeeper用于協調管理并保存Kafka集群的所有元數據信息，例如集群中有哪些Broker、Topic、分區等信息。

針對zookeeper的相關參數：

• zookeeper.connect：用于指定zookeeper的客戶端地址，zk1:2181,zk2:2181，同樣可以配置多個zookeeper（集群）

注意，若有兩套Kafka集群共用一套zookeeper集群，可以使用chroot別名來配置，格式如下：

# kafka1 集群
zk1:2181,zk2:2181,zk3:2181/kafka1
# kafka2 集群
zk1:2181,zk2:2181,zk3:2181/kafka2

只需要在zookeeper集群的最后添加 / [kafka集群名] 即可。

---

針對Broker之間連接和通信的參數：

• listeners：用于告訴外部連接者需要通過什么協議訪問指定的主機名和端口開放的Kafka服務。
• advertised.listeners：對外發布的監聽器，即面向外網。如果客戶端僅僅在內網中使用，那么就不需要配置該參數。

每個監聽器對于的格式為若干個逗號分割的三元組：<協議名稱，主機名，端口號>，例如：

CONTROLLER: //localhost:9092

---

針對Topic的相關參數：

• auto.create.topics.enable：是否允許自動創建Topic。最好設置為false，避免因為操作失誤創建錯誤名稱的主題。
• unclean.leader.election.enable：是否允許Unclean Leader選舉。
• auto.leader.rebalance.enable：是否允許定期進行Leader選舉（在滿足一定的條件下）。建議將其改成false，因為將其設置為true，會導致一個正常的Leader副本被其他的副本替換。本質上的性能收益幾乎沒有，反而替換一次的Leader代價卻很高。因此生產環境中建議設置為false。

關于Unclean Leader選舉：

背景：Kafka中，每個分區都有多個副本來提供高可用和備份，但是只有一個副本能作為Leader。因此副本之間需要選舉出一個Leader來，但并非所有副本都具有資格參與競選，只有保存數據比較多的副本才有資格。

那么當擁有競選資格的副本都掛了的前提下，此時還需不需要進行Leader選舉？ 該參數用于這樣的情形：

• unclean.leader.election.enable = false：即使發生上述情況，依舊不會競選新Leader。后果：該分區不可用。
• unclean.leader.election.enable = true：若發生上述情況，此時Kafka允許從落后的副本中競選出新副本。后果：可能有數據丟失的情況（落后副本本身保存的數據就不全）

---

針對數據留存的相關參數：

• log.retention.{hour|minutes|ms}：控制一條消息數據能夠被保存多長的時間，優先級：ms > minutes > hour。
• log.retention.bytes：指定Broker為消息保存的總磁盤容量大小。默認值為 -1 ，意思是條件允許的情況下，可以在該Broker上保存任意量的數據。一般來說使用場景在于：限制某個用戶使用的磁盤空間。
• message.max.bytes：控制Broker能夠接收的最大消息大小。默認1000012 bytes，還不到1MB，實際生產上的消息大小往往可能會超過1MB。

注意：上述提及的所有參數，都不要使用默認值的參數，最好要自己手動配置。
注意：上述提及的所有參數，都不要使用默認值的參數，最好要自己手動配置。
注意：上述提及的所有參數，都不要使用默認值的參數，最好要自己手動配置。

2.3.2 Topic級別參數

前提：Topic級別參數的優先級 > Broker端參數（全局）

重要的三個參數（和上面全局的配置非常相似）：

• retention.ms：規定了該Topic消息被保存的時長，默認7天，可覆蓋全局值。
• retention.bytes：規定了需要為該Topic預留多大的磁盤空間。
• max.message.bytes：能夠正常接收該Topic的最大消息大小。

在創建Topic的時候，可以通過以下命令來執行：

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic transaction \
--partitions 1 --replication-factor 1 --config retention.ms=15552000000 \
--config max.message.bytes=5242880

在修改Topic的時候，可以通過以下命令來執行：

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics \
--entity-name transaction --alter --add-config max.message.bytes=10485760

2.3.3 JVM參數

參考配置如下：

# 先設置對于的環境變量
export KAFKA_HEAP_OPTS=--Xms6g --Xmx6g
export KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true
# 啟動Kafka
bin/kafka-server-start.sh config/server.properties

2.3.4 操作系統參數

對Kafka中影響相對較大的OS參數有四個：

• 文件描述符限制。
• 文件系統類型：指的是諸如ext3、ext4或者XFS這樣的日志文件系統，生產上推薦使用XFS。
• Swappiness：將其設置為一個較小的值，倘若設置為0，當物理內存耗盡時，操作系統會觸發 OOM killer 這個組件，它會隨機挑選一個進程然后將其停止，不會留下任何的預警。
• 提交時間。

重點說下文件描述符限制。文件描述符限制實際上調大此值并不會造成什么嚴重的影響。當我們遇到這樣的報錯信息：Too many open files，就說明咱們的文件描述符限制太小了，需要將其調大。例如：

ulimit -n 1000000

然后說下提交時間（Flush刷新落盤時間）。

首先kafka中，何時認為數據已經寫入成功？只要數據被寫入到操作系統中的頁緩存（Page Cache）上就可以了，即內核緩沖區中的一部分。而并不是說等數據被寫入磁盤才認定為成功。

寫入到頁緩存中的數據，操作系統會根據LRU算法定期地將其寫入到物理磁盤上，而這個定期就是由提交時間來確定的，默認是5秒，一般情況下，我們可以適當的正常這個時間。

哪怕頁緩存中的數據在寫入到磁盤之前宕機了，發生了數據的丟失，但是鑒于Kafka的多副本機制，對于數據丟失的情況有所改善，因此這里可以稍微增加點提交時間，提升性能也是可以的。