今天，我將圍繞如下幾個問題進行分享：

• 為什麼需要消息系統?
• Kafka 架構原理?
• Kafka 如何存儲消息?
• Producer 如何發送消息?
• Consumer 如何消費消息?
• Offset 如何保存?
• 消息系統可能遇到哪些問題？

為什麼需要消息系統？

削峰

數據庫的處理能力是有限的，在峰值期，過多的請求落到後臺，一旦超過系統的處理能力，可能會使系統掛掉。

如上圖所示，系統的處理能力是 2k/s，MQ 處理能力是 8k/s，峰值請求 5k/s，MQ 的處理能力遠遠大於數據庫，在高峰期，請求可以先積壓在 MQ 中，系統可以根據自身的處理能力以 2k/s 的速度消費這些請求。

這樣等高峰期一過，請求可能只有 100/s，系統可以很快的消費掉積壓在 MQ 中的請求。

注意，上面的請求指的是寫請求，查詢請求一般通過緩存解決。

解耦

如下場景，S 系統與 A、B、C 系統緊密耦合。由於需求變動，A 系統修改了相關代碼，S 系統也需要調整 A 相關的代碼。

過幾天，C 系統需要刪除，S 緊跟著刪除 C 相關代碼；又過了幾天，需要新增 D 系統，S 系統又要添加與 D 相關的代碼；再過幾天，程序猿瘋了...

這樣各個系統緊密耦合，不利於維護，也不利於擴展。現在引入 MQ，A 系統變動，A 自己修改自己的代碼即可；C 系統刪除，直接取消訂閱；D 系統新增，訂閱相關消息即可。

這樣通過引入消息中間件，使各個系統都與 MQ 交互，從而避免它們之間的錯綜複雜的調用關係。

Kafka 架構原理?

Kafka 相關概念：

• Broker：Kafka 集群中包含的服務器。
• Producer：消息生產者。
• Consumer：消息消費者。
• Consumer Group：每個 Consumer 都屬於一個 Consumer Group，每條消息只能被 Consumer Group 中的一個 Consumer 消費，但可以被多個 Consumer Group 消費。
• Topic：消息的類別。每條消息都屬於某個 Topic，不同的 Topic 之間是相互獨立的，即 Kafka 是面向 Topic 的。
• Partition：每個 Topic 分為多個 Partition，Partition 是 Kafka 分配的單位。Kafka 物理上的概念，相當於一個目錄，目錄下的日誌文件構成這個 Partition。
• Replica：Partition 的副本，保障 Partition 的高可用。
• Leader：Replica 中的一個角色， Producer 和 Consumer 只跟 Leader 交互。
• Follower：Replica 中的一個角色，從 Leader 中複製數據。
• Controller：Kafka 集群中的其中一個服務器，用來進行 Leader Election 以及各種 Failover。
• Zookeeper：Kafka 通過 Zookeeper 來存儲集群的 Meta 信息。

Topic and Logs

Message 是按照 Topic 來組織的，每個 Topic 可以分成多個 Partition（對應 server.properties/num.partitions）。

Partition 是一個順序的追加日誌，屬於順序寫磁盤（順序寫磁盤效率比隨機寫內存要高，保障 Kafka 吞吐率）。

其結構如下：server.properties/num.partitions 表示文件 server.properties 中的 num.partitions 配置項，下同。

Partition 中的每條記錄（Message）包含三個屬性：Offset，messageSize 和 Data。

其中 Offset 表示消息偏移量；messageSize 表示消息的大小；Data 表示消息的具體內容。

Partition 是以文件的形式存儲在文件系統中，位置由 server.properties/log.dirs 指定，其命名規則為 <topic_name>-<partition_id>。

比如，Topic 為"page_visits"的消息，分為 5 個 Partition，其目錄結構為：

Partition 可能位於不同的 Broker 上，Partition 是分段的，每個段是一個 Segment 文件。

Segment的常用配置有：

#server.properties
#segment文件的大小，默認為 1G
log.segment.bytes=1024*1024*1024
#滾動生成新的segment文件的最大時長
log.roll.hours=24*7
#segment文件保留的最大時長，超時將被刪除
log.retention.hours=24*7

Partition 目錄下包括了數據文件和索引文件，下圖是某個 Partition 的目錄結構：

Index 採用稀疏存儲的方式，它不會為每一條 Message 都建立索引，而是每隔一定的字節數建立一條索引，避免索引文件佔用過多的空間。

缺點是沒有建立索引的 Offset 不能一次定位到 Message 的位置，需要做一次順序掃描，但是掃描的範圍很小。

索引包含兩個部分（均為 4 個字節的數字），分別為相對 Offset 和 Position。

相對 Offset 表示 Segment 文件中的 Offset，Position 表示 Message 在數據文件中的位置。

總結：Kafka 的 Message 存儲採用了分區（Partition），磁盤順序讀寫，分段（LogSegment）和稀疏索引這幾個手段來達到高效性。

Partition and Replica

一個 Topic 物理上分為多個 Partition，位於不同的 Broker 上。如果沒有 Replica，一旦 Broker 宕機，其上所有的 Patition 將不可用。

每個 Partition 可以有多個Replica（對應server.properties/default.replication.factor），分配到不同的 Broker 上。

其中有一個 Leader 負責讀寫，處理來自 Producer 和 Consumer 的請求；其他作為 Follower 從 Leader Pull 消息，保持與 Leader 的同步。

如何分配 Partition 和 Replica 到 Broker 上？步驟如下：

• 將所有 Broker（假設共 n 個 Broker）和待分配的 Partition 排序。
• 將第 i 個 Partition 分配到第（i mod n）個 Broker 上。
• 將第 i 個 Partition 的第 j 個 Replica 分配到第（(i + j) mode n）個 Broker 上。

根據上面的分配規則，若 Replica 的數量大於 Broker 的數量，必定會有兩個相同的 Replica 分配到同一個 Broker 上，產生冗餘。因此 Replica 的數量應該小於或等於 Broker 的數量。

Leader 選舉

Kafka 在 Zookeeper 中（/brokers/topics/[topic]/partitions/[partition]/state）動態維護了一個 ISR（in-sync replicas）。

ISR 裡面的所有 Replica 都"跟上"了 Leader，Controller 將會從 ISR 裡選一個做 Leader。

具體流程如下：

• Controller 在 Zookeeper 的 /brokers/ids/[brokerId] 節點註冊 Watcher，當 Broker 宕機時 Zookeeper 會 Fire Watch。
• Controller 從 /brokers/ids 節點讀取可用 Broker。
• Controller 決定 set_p，該集合包含宕機 Broker 上的所有 Partition。
• 對 set_p 中的每一個 Partition，從/brokers/topics/[topic]/partitions/[partition]/state 節點讀取 ISR，決定新 Leader，將新 Leader、ISR、controller_epoch 和 leader_epoch 等信息寫入 State 節點。
• 通過 RPC 向相關 Broker 發送 leaderAndISRRequest 命令。

當 ISR 為空時，會選一個 Replica（不一定是 ISR 成員）作為 Leader；當所有的 Replica 都歇菜了，會等任意一個 Replica 復活，將其作為 Leader。

ISR（同步列表）中的 Follower 都"跟上"了Leader，"跟上"並不表示完全一致，它由 server.properties/replica.lag.time.max.ms 配置。

表示 Leader 等待 Follower 同步消息的最大時間，如果超時，Leader 將 Follower 移除 ISR。配置項 replica.lag.max.messages 已經移除。

Replica 同步

Kafka 通過"拉模式"同步消息，即 Follower 從 Leader 批量拉取數據來同步。

具體的可靠性，是由生產者（根據配置項 producer.properties/acks）來決定的。

In Kafka 0.9，request.required.acks=-1 which configration of producer is replaced by acks=all, but this old config is remained in docs.

在 0.9 版本，生產者配置項 request.required.acks=-1 被 acks=all 取代，但是老的配置項還保留在文檔中。

PS：最新的文檔 2.2.x request.required.acks 已經不存在了。

在 Acks=-1 的時候，如果 ISR 少於 min.insync.replicas 指定的數目，將會拋出 NotEnoughReplicas 或 NotEnoughReplicasAfterAppend 異常。

Producer 如何發送消息？

Producer 首先將消息封裝進一個 ProducerRecord 實例中。

消息路由：

• 發送消息時如果指定了 Partition，則直接使用。
• 如果指定了 Key，則對 Key 進行哈希，選出一個 Partition。這個 Hash（即分區機制）由 producer.properties/partitioner.class 指定的類實現，這個路由類需要實現 Partitioner 接口。
• 如果都未指定，通過 Round-Robin 來選 Partition。

消息並不會立即發送，而是先進行序列化後，發送給 Partitioner，也就是上面提到的 Hash 函數，由 Partitioner 確定目標分區後，發送到一塊內存緩衝區中（發送隊列）。

Producer 的另一個工作線程（即 Sender 線程），則負責實時地從該緩衝區中提取出準備好的消息封裝到一個批次內，統一發送到對應的 Broker 中。

其過程大致是這樣的：

Consumer 如何消費消息？

每個 Consumer 都劃歸到一個邏輯 Consumer Group 中，一個 Partition 只能被同一個 Consumer Group 中的一個 Consumer 消費，但可以被不同的 Consumer Group 消費。

若 Topic 的 Partition 數量為 p，Consumer Group 中訂閱此 Topic 的 Consumer 數量為 c， 則：

p < c: 會有 c - p 個 consumer閒置，造成浪費
p > c: 一個 consumer 對應多個 partition 
p = c: 一個 consumer 對應一個 partition

應該合理分配 Consumer 和 Partition 的數量，避免造成資源傾斜，最好 Partiton 數目是 Consumer 數目的整數倍。

①如何將 Partition 分配給 Consumer

生產過程中 Broker 要分配 Partition，消費過程這裡，也要分配 Partition 給消費者。

類似 Broker 中選了一個 Controller 出來，消費也要從 Broker 中選一個 Coordinator，用於分配 Partition。

當 Partition 或 Consumer 數量發生變化時，比如增加 Consumer，減少 Consumer（主動或被動），增加 Partition，都會進行 Rebalance。

其過程如下：

• Consumer 給 Coordinator 發送 JoinGroupRequest 請求。這時其他 Consumer 發 Heartbeat 請求過來時，Coordinator 會告訴他們，要 Rebalance了。其他 Consumer 也發送 JoinGroupRequest 請求。
• Coordinator 在 Consumer 中選出一個 Leader，其他作為 Follower，通知給各個 Consumer，對於 Leader，還會把 Follower 的 Metadata 帶給它。
• Consumer Leader 根據 Consumer Metadata 重新分配 Partition。
• Consumer 向 Coordinator 發送 SyncGroupRequest，其中 Leader 的 SyncGroupRequest 會包含分配的情況。Coordinator 回包，把分配的情況告訴 Consumer，包括 Leader。

②Consumer Fetch Message

Consumer 採用"拉模式"消費消息，這樣 Consumer 可以自行決定消費的行為。

Consumer 調用 Poll（duration）從服務器拉取消息。拉取消息的具體行為由下面的配置項決定：

#consumer.properties
#消費者最多 poll 多少個 record
max.poll.records=500
#消費者 poll 時 partition 返回的最大數據量
max.partition.fetch.bytes=1048576
#Consumer 最大 poll 間隔
#超過此值服務器會認為此 consumer failed 
#並將此 consumer 踢出對應的 consumer group 
max.poll.interval.ms=300000

在 Partition 中，每個消息都有一個 Offset。新消息會被寫到 Partition 末尾（最新的一個 Segment 文件末尾）， 每個 Partition 上的消息是順序消費的，不同的 Partition 之間消息的消費順序是不確定的。

若一個 Consumer 消費多個 Partition, 則各個 Partition 之前消費順序是不確定的，但在每個 Partition 上是順序消費。

若來自不同 Consumer Group 的多個 Consumer 消費同一個 Partition，則各個 Consumer 之間的消費互不影響，每個 Consumer 都會有自己的 Offset。

Consumer A 和 Consumer B 屬於不同的 Consumer Group。Cosumer A 讀取到 Offset=9， Consumer B 讀取到 Offset=11，這個值表示下次讀取的位置。

也就是說 Consumer A 已經讀取了 Offset 為 0~8 的消息，Consumer B 已經讀取了 Offset 為 0～10 的消息。

下次從 Offset=9 開始讀取的 Consumer 並不一定還是 Consumer A 因為可能發生 Rebalance。

Offset 如何保存？

Consumer 消費 Partition 時，需要保存 Offset 記錄當前消費位置。

Offset 可以選擇自動提交或調用 Consumer 的 commitSync() 或 commitAsync() 手動提交，相關配置為：

#是否自動提交 offset
enable.auto.commit=true
#自動提交間隔。enable.auto.commit=true 時有效
auto.commit.interval.ms=5000

Offset 保存在名叫 __consumeroffsets 的 Topic 中。寫消息的 Key 由 GroupId、Topic、Partition 組成，Value 是 Offset。

一般情況下，每個 Key 的 Offset 都是緩存在內存中，查詢的時候不用遍歷 Partition，如果沒有緩存，第一次就會遍歷 Partition 建立緩存，然後查詢返回。

__consumeroffsets 的 Partition 數量由下面的 Server 配置決定：

offsets.topic.num.partitions=50

Offset 保存在哪個分區上，即 __consumeroffsets 的分區機制，可以表示為：

groupId.hashCode() mode groupMetadataTopicPartitionCount

groupMetadataTopicPartitionCount 是上面配置的分區數。因為一個 Partition 只能被同一個 Consumer Group 的一個 Consumer 消費，因此可以用 GroupId 表示此 Consumer 消費 Offeset 所在分區。

消息系統可能遇到哪些問題？

Kafka 支持 3 種消息投遞語義：

• at most once：最多一次，消息可能會丟失，但不會重複
• 獲取數據 -> commit offset -> 業務處理
• at least once：最少一次，消息不會丟失，可能會重複
• 獲取數據 -> 業務處理 -> commit offset。
• exactly once：只且一次，消息不丟失不重複，只且消費一次（0.11 中實現，僅限於下游也是 Kafka）

①如何保證消息不被重複消費？（消息的冪等性）

對於更新操作，天然具有冪等性。對於新增操作，可以給每條消息一個唯一的 id，處理前判斷是否被處理過。這個 id 可以存儲在 Redis 中，如果是寫數據庫可以用主鍵約束。

②如何保證消息的可靠性傳輸？（消息丟失的問題）

根據 Kafka 架構，有三個地方可能丟失消息：Consumer，Producer 和 Server。

消費端弄丟了數據：當 server.properties/enable.auto.commit 設置為 True 的時候，Kafka 會先 Commit Offset 再處理消息，如果這時候出現異常，這條消息就丟失了。

因此可以關閉自動提交 Offset，在處理完成後手動提交 Offset，這樣可以保證消息不丟失；但是如果提交 Offset 失敗，可能導致重複消費的問題， 這時保證冪等性即可。

Kafka 弄丟了消息：如果某個 Broker 不小心掛了，此時若 Replica 只有一個，Broker 上的消息就丟失了。

若 Replica>1，給 Leader 重新選一個 Follower 作為新的 Leader，如果 Follower 還有些消息沒有同步，這部分消息便丟失了。

可以進行如下配置，避免上面的問題：

• 給 Topic 設置 replication.factor 參數：這個值必須大於 1，要求每個 Partition 必須有至少 2 個副本。
• 在 Kafka 服務端設置 min.insync.replicas 參數：這個值必須大於 1，這個是要求一個 Leader 至少感知到有至少一個 Follower 還跟自己保持聯繫，沒掉隊，這樣才能確保 Leader 掛了還有一個 Follower 吧。
• 在 Producer 端設置 acks=all：這個是要求每條數據，必須是寫入所有 Replica 之後，才能認為是寫成功了。
• 在 Producer 端設置 retries=MAX（很大很大很大的一個值，無限次重試的意思）：這個是要求一旦寫入失敗，就無限重試，卡在這裡了。

Producer弄丟了消息：在 Producer 端設置 acks=all，保證所有的 ISR 都同步了消息才認為寫入成功。

③如何保證消息的順序性？

Kafka 中 Partition 上的消息是順序的，可以將需要順序消費的消息發送到同一個 Partition 上，用單個 Consumer 消費。

上面是學習 Kafka 時總結的，如有錯誤或不合理的地方，歡迎指正！

作者：lbzhello

轉載於：https://mp.weixin.qq.com/s/ZXlW3dp0Y05UFwgr7izqMQ