什么是消息隊(duì)列?簡(jiǎn)單來(lái)說(shuō)，消息隊(duì)列是存放消息的容器。客戶端可以將消息發(fā)送到消息服務(wù)器，也可以從消息服務(wù)器獲取消息。

[[270989]]

圖片來(lái)自 Pexels

今天，我將圍繞如下幾個(gè)問(wèn)題進(jìn)行分享：

• 為什么需要消息系統(tǒng)?
• Kafka 架構(gòu)原理?
• Kafka 如何存儲(chǔ)消息?
• Producer 如何發(fā)送消息?
• Consumer 如何消費(fèi)消息?
• Offset 如何保存?
• 消息系統(tǒng)可能遇到哪些問(wèn)題?

為什么需要消息系統(tǒng)?

削峰

數(shù)據(jù)庫(kù)的處理能力是有限的，在峰值期，過(guò)多的請(qǐng)求落到后臺(tái)，一旦超過(guò)系統(tǒng)的處理能力，可能會(huì)使系統(tǒng)掛掉。

如上圖所示，系統(tǒng)的處理能力是 2k/s，MQ 處理能力是 8k/s，峰值請(qǐng)求 5k/s，MQ 的處理能力遠(yuǎn)遠(yuǎn)大于數(shù)據(jù)庫(kù)，在高峰期，請(qǐng)求可以先積壓在 MQ 中，系統(tǒng)可以根據(jù)自身的處理能力以 2k/s 的速度消費(fèi)這些請(qǐng)求。

這樣等高峰期一過(guò)，請(qǐng)求可能只有 100/s，系統(tǒng)可以很快的消費(fèi)掉積壓在 MQ 中的請(qǐng)求。

注意，上面的請(qǐng)求指的是寫請(qǐng)求，查詢請(qǐng)求一般通過(guò)緩存解決。

解耦

如下場(chǎng)景，S 系統(tǒng)與 A、B、C 系統(tǒng)緊密耦合。由于需求變動(dòng)，A 系統(tǒng)修改了相關(guān)代碼，S 系統(tǒng)也需要調(diào)整 A 相關(guān)的代碼。

過(guò)幾天，C 系統(tǒng)需要?jiǎng)h除，S 緊跟著刪除 C 相關(guān)代碼;又過(guò)了幾天，需要新增 D 系統(tǒng)，S 系統(tǒng)又要添加與 D 相關(guān)的代碼;再過(guò)幾天，程序猿瘋了...

這樣各個(gè)系統(tǒng)緊密耦合，不利于維護(hù)，也不利于擴(kuò)展。現(xiàn)在引入 MQ，A 系統(tǒng)變動(dòng)，A 自己修改自己的代碼即可;C 系統(tǒng)刪除，直接取消訂閱;D 系統(tǒng)新增，訂閱相關(guān)消息即可。

這樣通過(guò)引入消息中間件，使各個(gè)系統(tǒng)都與 MQ 交互，從而避免它們之間的錯(cuò)綜復(fù)雜的調(diào)用關(guān)系。

Kafka 架構(gòu)原理?

Kafka 相關(guān)概念：

• Broker：Kafka 集群中包含的服務(wù)器。
• Producer：消息生產(chǎn)者。
• Consumer：消息消費(fèi)者。
• Consumer Group：每個(gè) Consumer 都屬于一個(gè) Consumer Group，每條消息只能被 Consumer Group 中的一個(gè) Consumer 消費(fèi)，但可以被多個(gè) Consumer Group 消費(fèi)。
• Topic：消息的類別。每條消息都屬于某個(gè) Topic，不同的 Topic 之間是相互獨(dú)立的，即 Kafka 是面向 Topic 的。
• Partition：每個(gè) Topic 分為多個(gè) Partition，Partition 是 Kafka 分配的單位。Kafka 物理上的概念，相當(dāng)于一個(gè)目錄，目錄下的日志文件構(gòu)成這個(gè) Partition。
• Replica：Partition 的副本，保障 Partition 的高可用。
• Leader：Replica 中的一個(gè)角色， Producer 和 Consumer 只跟 Leader 交互。
• Follower：Replica 中的一個(gè)角色，從 Leader 中復(fù)制數(shù)據(jù)。
• Controller：Kafka 集群中的其中一個(gè)服務(wù)器，用來(lái)進(jìn)行 Leader Election 以及各種 Failover。
• Zookeeper：Kafka 通過(guò) Zookeeper 來(lái)存儲(chǔ)集群的 Meta 信息。

Topic and Logs

Message 是按照 Topic 來(lái)組織的，每個(gè) Topic 可以分成多個(gè) Partition(對(duì)應(yīng) server.properties/num.partitions)。

Partition 是一個(gè)順序的追加日志，屬于順序?qū)懘疟P(順序?qū)懘疟P效率比隨機(jī)寫內(nèi)存要高，保障 Kafka 吞吐率)。

其結(jié)構(gòu)如下：server.properties/num.partitions 表示文件 server.properties 中的 num.partitions 配置項(xiàng)，下同。

Partition 中的每條記錄(Message)包含三個(gè)屬性：Offset，messageSize 和 Data。

其中 Offset 表示消息偏移量;messageSize 表示消息的大小;Data 表示消息的具體內(nèi)容。

Partition 是以文件的形式存儲(chǔ)在文件系統(tǒng)中，位置由 server.properties/log.dirs 指定，其命名規(guī)則為 - 。

比如，Topic 為"page_visits"的消息，分為 5 個(gè) Partition，其目錄結(jié)構(gòu)為：

Partition 可能位于不同的 Broker 上，Partition 是分段的，每個(gè)段是一個(gè) Segment 文件。

Segment的常用配置有：

#server.properties 
 
#segment文件的大小，默認(rèn)為 1G 
log.segment.bytes=1024*1024*1024 
#滾動(dòng)生成新的segment文件的最大時(shí)長(zhǎng) 
log.roll.hours=24*7 
#segment文件保留的最大時(shí)長(zhǎng)，超時(shí)將被刪除 
log.retention.hours=24*7 

Partition 目錄下包括了數(shù)據(jù)文件和索引文件，下圖是某個(gè) Partition 的目錄結(jié)構(gòu)：

Index 采用稀疏存儲(chǔ)的方式，它不會(huì)為每一條 Message 都建立索引，而是每隔一定的字節(jié)數(shù)建立一條索引，避免索引文件占用過(guò)多的空間。

缺點(diǎn)是沒(méi)有建立索引的 Offset 不能一次定位到 Message 的位置，需要做一次順序掃描，但是掃描的范圍很小。

索引包含兩個(gè)部分(均為 4 個(gè)字節(jié)的數(shù)字)，分別為相對(duì) Offset 和 Position。

相對(duì) Offset 表示 Segment 文件中的 Offset，Position 表示 Message 在數(shù)據(jù)文件中的位置。

總結(jié)：Kafka 的 Message 存儲(chǔ)采用了分區(qū)(Partition)，磁盤順序讀寫，分段(LogSegment)和稀疏索引這幾個(gè)手段來(lái)達(dá)到高效性。

Partition and Replica

一個(gè) Topic 物理上分為多個(gè) Partition，位于不同的 Broker 上。如果沒(méi)有 Replica，一旦 Broker 宕機(jī)，其上所有的 Patition 將不可用。

每個(gè) Partition 可以有多個(gè)Replica(對(duì)應(yīng)server.properties/default.replication.factor)，分配到不同的 Broker 上。

其中有一個(gè) Leader 負(fù)責(zé)讀寫，處理來(lái)自 Producer 和 Consumer 的請(qǐng)求;其他作為 Follower 從 Leader Pull 消息，保持與 Leader 的同步。

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

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

根據(jù)上面的分配規(guī)則，若 Replica 的數(shù)量大于 Broker 的數(shù)量，必定會(huì)有兩個(gè)相同的 Replica 分配到同一個(gè) Broker 上，產(chǎn)生冗余。因此 Replica 的數(shù)量應(yīng)該小于或等于 Broker 的數(shù)量。

Leader 選舉

Kafka 在 Zookeeper 中(/brokers/topics/[topic]/partitions/[partition]/state)動(dòng)態(tài)維護(hù)了一個(gè) ISR(in-sync replicas)。

ISR 里面的所有 Replica 都"跟上"了 Leader，Controller 將會(huì)從 ISR 里選一個(gè)做 Leader。

具體流程如下：

• Controller 在 Zookeeper 的 /brokers/ids/[brokerId] 節(jié)點(diǎn)注冊(cè) Watcher，當(dāng) Broker 宕機(jī)時(shí) Zookeeper 會(huì) Fire Watch。
• Controller 從 /brokers/ids 節(jié)點(diǎn)讀取可用 Broker。
• Controller 決定 set_p，該集合包含宕機(jī) Broker 上的所有 Partition。
• 對(duì) set_p 中的每一個(gè) Partition，從/brokers/topics/[topic]/partitions/[partition]/state 節(jié)點(diǎn)讀取 ISR，決定新 Leader，將新 Leader、ISR、controller_epoch 和 leader_epoch 等信息寫入 State 節(jié)點(diǎn)。
• 通過(guò) RPC 向相關(guān) Broker 發(fā)送 leaderAndISRRequest 命令。

當(dāng) ISR 為空時(shí)，會(huì)選一個(gè) Replica(不一定是 ISR 成員)作為 Leader;當(dāng)所有的 Replica 都歇菜了，會(huì)等任意一個(gè) Replica 復(fù)活，將其作為 Leader。

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

表示 Leader 等待 Follower 同步消息的最大時(shí)間，如果超時(shí)，Leader 將 Follower 移除 ISR。配置項(xiàng) replica.lag.max.messages 已經(jīng)移除。

Replica 同步

Kafka 通過(guò)"拉模式"同步消息，即 Follower 從 Leader 批量拉取數(shù)據(jù)來(lái)同步。

具體的可靠性，是由生產(chǎn)者(根據(jù)配置項(xiàng) producer.properties/acks)來(lái)決定的。

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 版本，生產(chǎn)者配置項(xiàng) request.required.acks=-1 被 acks=all 取代，但是老的配置項(xiàng)還保留在文檔中。

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

在 Acks=-1 的時(shí)候，如果 ISR 少于 min.insync.replicas 指定的數(shù)目，將會(huì)拋出 NotEnoughReplicas 或 NotEnoughReplicasAfterAppend 異常。

Producer 如何發(fā)送消息?

Producer 首先將消息封裝進(jìn)一個(gè) ProducerRecord 實(shí)例中。

消息路由：

• 發(fā)送消息時(shí)如果指定了 Partition，則直接使用。
• 如果指定了 Key，則對(duì) Key 進(jìn)行哈希，選出一個(gè) Partition。這個(gè) Hash(即分區(qū)機(jī)制)由 producer.properties/partitioner.class 指定的類實(shí)現(xiàn)，這個(gè)路由類需要實(shí)現(xiàn) Partitioner 接口。
• 如果都未指定，通過(guò) Round-Robin 來(lái)選 Partition。

消息并不會(huì)立即發(fā)送，而是先進(jìn)行序列化后，發(fā)送給 Partitioner，也就是上面提到的 Hash 函數(shù)，由 Partitioner 確定目標(biāo)分區(qū)后，發(fā)送到一塊內(nèi)存緩沖區(qū)中(發(fā)送隊(duì)列)。

Producer 的另一個(gè)工作線程(即 Sender 線程)，則負(fù)責(zé)實(shí)時(shí)地從該緩沖區(qū)中提取出準(zhǔn)備好的消息封裝到一個(gè)批次內(nèi)，統(tǒng)一發(fā)送到對(duì)應(yīng)的 Broker 中。

其過(guò)程大致是這樣的：

圖片來(lái)自 123archu

Consumer 如何消費(fèi)消息?

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

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

p < c: 會(huì)有 c - p 個(gè) consumer閑置，造成浪費(fèi) 
p > c: 一個(gè) consumer 對(duì)應(yīng)多個(gè) partition  
p = c: 一個(gè) consumer 對(duì)應(yīng)一個(gè) partition 

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

①如何將 Partition 分配給 Consumer

生產(chǎn)過(guò)程中 Broker 要分配 Partition，消費(fèi)過(guò)程這里，也要分配 Partition 給消費(fèi)者。

類似 Broker 中選了一個(gè) Controller 出來(lái)，消費(fèi)也要從 Broker 中選一個(gè) Coordinator，用于分配 Partition。

當(dāng) Partition 或 Consumer 數(shù)量發(fā)生變化時(shí)，比如增加 Consumer，減少 Consumer(主動(dòng)或被動(dòng))，增加 Partition，都會(huì)進(jìn)行 Rebalance。

其過(guò)程如下：

• Consumer 給 Coordinator 發(fā)送 JoinGroupRequest 請(qǐng)求。這時(shí)其他 Consumer 發(fā) Heartbeat 請(qǐng)求過(guò)來(lái)時(shí)，Coordinator 會(huì)告訴他們，要 Rebalance了。其他 Consumer 也發(fā)送 JoinGroupRequest 請(qǐng)求。
• Coordinator 在 Consumer 中選出一個(gè) Leader，其他作為 Follower，通知給各個(gè) Consumer，對(duì)于 Leader，還會(huì)把 Follower 的 Metadata 帶給它。
• Consumer Leader 根據(jù) Consumer Metadata 重新分配 Partition。
• Consumer 向 Coordinator 發(fā)送 SyncGroupRequest，其中 Leader 的 SyncGroupRequest 會(huì)包含分配的情況。Coordinator 回包，把分配的情況告訴 Consumer，包括 Leader。

②Consumer Fetch Message

Consumer 采用"拉模式"消費(fèi)消息，這樣 Consumer 可以自行決定消費(fèi)的行為。

Consumer 調(diào)用 Poll(duration)從服務(wù)器拉取消息。拉取消息的具體行為由下面的配置項(xiàng)決定：

#consumer.properties 
 
#消費(fèi)者最多 poll 多少個(gè) record 
max.poll.records=500 
 
#消費(fèi)者 poll 時(shí) partition 返回的最大數(shù)據(jù)量 
max.partition.fetch.bytes=1048576 
 
#Consumer 最大 poll 間隔 
#超過(guò)此值服務(wù)器會(huì)認(rèn)為此 consumer failed  
#并將此 consumer 踢出對(duì)應(yīng)的 consumer group  
max.poll.interval.ms=300000 

在 Partition 中，每個(gè)消息都有一個(gè) Offset。新消息會(huì)被寫到 Partition 末尾(最新的一個(gè) Segment 文件末尾)， 每個(gè) Partition 上的消息是順序消費(fèi)的，不同的 Partition 之間消息的消費(fèi)順序是不確定的。

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

若來(lái)自不同 Consumer Group 的多個(gè) Consumer 消費(fèi)同一個(gè) Partition，則各個(gè) Consumer 之間的消費(fèi)互不影響，每個(gè) Consumer 都會(huì)有自己的 Offset。

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

也就是說(shuō) Consumer A 已經(jīng)讀取了 Offset 為 0~8 的消息，Consumer B 已經(jīng)讀取了 Offset 為 0～10 的消息。

下次從 Offset=9 開(kāi)始讀取的 Consumer 并不一定還是 Consumer A 因?yàn)榭赡馨l(fā)生 Rebalance。

Offset 如何保存?

Consumer 消費(fèi) Partition 時(shí)，需要保存 Offset 記錄當(dāng)前消費(fèi)位置。

Offset 可以選擇自動(dòng)提交或調(diào)用 Consumer 的 commitSync() 或 commitAsync() 手動(dòng)提交，相關(guān)配置為：

#是否自動(dòng)提交 offset 
enable.auto.commit=true 
 
#自動(dòng)提交間隔。enable.auto.commit=true 時(shí)有效 
auto.commit.interval.ms=5000 

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

一般情況下，每個(gè) Key 的 Offset 都是緩存在內(nèi)存中，查詢的時(shí)候不用遍歷 Partition，如果沒(méi)有緩存，第一次就會(huì)遍歷 Partition 建立緩存，然后查詢返回。

__consumeroffsets 的 Partition 數(shù)量由下面的 Server 配置決定：

offsets.topic.num.partitions=50 

Offset 保存在哪個(gè)分區(qū)上，即 __consumeroffsets 的分區(qū)機(jī)制，可以表示為：

groupId.hashCode() mode groupMetadataTopicPartitionCount 

groupMetadataTopicPartitionCount 是上面配置的分區(qū)數(shù)。因?yàn)橐粋€(gè) Partition 只能被同一個(gè) Consumer Group 的一個(gè) Consumer 消費(fèi)，因此可以用 GroupId 表示此 Consumer 消費(fèi) Offeset 所在分區(qū)。

消息系統(tǒng)可能遇到哪些問(wèn)題?

Kafka 支持 3 種消息投遞語(yǔ)義：

• at most once：最多一次，消息可能會(huì)丟失，但不會(huì)重復(fù)

獲取數(shù)據(jù) -> commit offset -> 業(yè)務(wù)處理

• at least once：最少一次，消息不會(huì)丟失，可能會(huì)重復(fù)

獲取數(shù)據(jù) -> 業(yè)務(wù)處理 -> commit offset。

• exactly once：只且一次，消息不丟失不重復(fù)，只且消費(fèi)一次(0.11 中實(shí)現(xiàn)，僅限于下游也是 Kafka)

①如何保證消息不被重復(fù)消費(fèi)?(消息的冪等性)

對(duì)于更新操作，天然具有冪等性。對(duì)于新增操作，可以給每條消息一個(gè)唯一的 id，處理前判斷是否被處理過(guò)。這個(gè) id 可以存儲(chǔ)在 Redis 中，如果是寫數(shù)據(jù)庫(kù)可以用主鍵約束。

②如何保證消息的可靠性傳輸?(消息丟失的問(wèn)題)

根據(jù) Kafka 架構(gòu)，有三個(gè)地方可能丟失消息：Consumer，Producer 和 Server。

消費(fèi)端弄丟了數(shù)據(jù)：當(dāng) server.properties/enable.auto.commit 設(shè)置為 True 的時(shí)候，Kafka 會(huì)先 Commit Offset 再處理消息，如果這時(shí)候出現(xiàn)異常，這條消息就丟失了。

因此可以關(guān)閉自動(dòng)提交 Offset，在處理完成后手動(dòng)提交 Offset，這樣可以保證消息不丟失;但是如果提交 Offset 失敗，可能導(dǎo)致重復(fù)消費(fèi)的問(wèn)題， 這時(shí)保證冪等性即可。

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

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

可以進(jìn)行如下配置，避免上面的問(wèn)題：

• 給 Topic 設(shè)置 replication.factor 參數(shù)：這個(gè)值必須大于 1，要求每個(gè) Partition 必須有至少 2 個(gè)副本。
• 在 Kafka 服務(wù)端設(shè)置 min.insync.replicas 參數(shù)：這個(gè)值必須大于 1，這個(gè)是要求一個(gè) Leader 至少感知到有至少一個(gè) Follower 還跟自己保持聯(lián)系，沒(méi)掉隊(duì)，這樣才能確保 Leader 掛了還有一個(gè) Follower 吧。
• 在 Producer 端設(shè)置 acks=all：這個(gè)是要求每條數(shù)據(jù)，必須是寫入所有 Replica 之后，才能認(rèn)為是寫成功了。
• 在 Producer 端設(shè)置 retries=MAX(很大很大很大的一個(gè)值，無(wú)限次重試的意思)：這個(gè)是要求一旦寫入失敗，就無(wú)限重試，卡在這里了。

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

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

Kafka 中 Partition 上的消息是順序的，可以將需要順序消費(fèi)的消息發(fā)送到同一個(gè) Partition 上，用單個(gè) Consumer 消費(fèi)。

上面是學(xué)習(xí) Kafka 時(shí)總結(jié)的，如有錯(cuò)誤或不合理的地方，歡迎指正!

參考資料：

• kafka 學(xué)習(xí)筆記：知識(shí)點(diǎn)整理
• advanced-java
• Kafka 的 Log 存儲(chǔ)解析
• kafka 生產(chǎn)者 Producer 參數(shù)設(shè)置及參數(shù)調(diào)優(yōu)建議-商業(yè)環(huán)境實(shí)戰(zhàn)系列
• 震驚了!原來(lái)這才是 kafka!
• kafka configuration
• kafka 2.3.0 API
• kafka consumer 配置詳解和提交方式

作者：lbzhello

簡(jiǎn)介：Java 程序猿，郵箱：lbzhello@qq.com