1. 指明 partition 的情況下，直接將指明的值直接作為 partiton 值;
2. 沒(méi)有指明 partition 值但有 key 的情況下，將 key 的 hash 值與 topic 的 partition 數(shù)進(jìn)行取余得到 partition 值;
3. 既沒(méi)有 partition 值又沒(méi)有 key 值的情況下，第一次調(diào)用時(shí)隨機(jī)生成一個(gè)整數(shù)(后面每次調(diào)用在這個(gè)整數(shù)上自增)，將這個(gè)值與 topic 可用的 partition 總數(shù)取余得到 partition 值，也就是常說(shuō)的 round-robin 算法。

2.3 副本(Replication)

同一個(gè) partition 可能會(huì)有多個(gè) replication( 對(duì)應(yīng) server.properties 配置中的 default.replication.factor=N)。沒(méi)有 replication 的情況下，一旦 broker 宕機(jī)，其上所有 patition 的數(shù)據(jù)都不可被消費(fèi)，同時(shí) producer 也不能再將數(shù)據(jù)存于其上的 patition。引入 replication 之后，同一個(gè) partition 可能會(huì)有多個(gè) replication，而這時(shí)需要在這些 replication 之間選出一 個(gè) leader， producer 和 consumer 只與這個(gè) leader 交互，其它 replication 作為 follower 從 leader 中復(fù)制數(shù)據(jù)。

2.4 數(shù)據(jù)可靠性保證

為保證 producer 發(fā)送的數(shù)據(jù)，能可靠的發(fā)送到指定的 topic，topic 的每個(gè) partition 收到 producer 數(shù)據(jù)后，都需要向 producer 發(fā)送 ack(acknowledgement確認(rèn)收到)，如果 producer 收到 ack，就會(huì)進(jìn)行下一輪的發(fā)送，否則重新發(fā)送數(shù)據(jù)。

a) 副本數(shù)據(jù)同步策略主要有如下兩種

Kafka 選擇了第二種方案，原因如下：

• 同樣為了容忍 n 臺(tái)節(jié)點(diǎn)的故障，第一種方案需要的副本數(shù)相對(duì)較多，而 Kafka 的每個(gè)分區(qū)都有大量的數(shù)據(jù)，第一種方案會(huì)造成大量的數(shù)據(jù)冗余;
• 雖然第二種方案的網(wǎng)絡(luò)延遲會(huì)比較高，但網(wǎng)絡(luò)延遲對(duì) Kafka 的影響較小。

b) ISR

采用第二種方案之后，設(shè)想一下情景：leader 收到數(shù)據(jù)，所有 follower 都開(kāi)始同步數(shù)據(jù)，但有一個(gè) follower 掛了，遲遲不能與 leader 保持同步，那 leader 就要一直等下去，直到它完成同步，才能發(fā)送 ack，這個(gè)問(wèn)題怎么解決呢?

leader 維護(hù)了一個(gè)動(dòng)態(tài)的 in-sync replica set(ISR)，意為和 leader 保持同步的 follower 集合。當(dāng) ISR 中的follower 完成數(shù)據(jù)的同步之后，leader 就會(huì)給 follower 發(fā)送 ack。如果 follower 長(zhǎng)時(shí)間未向 leader 同步數(shù)據(jù)，則該 follower 將會(huì)被踢出 ISR，該時(shí)間閾值由 replica.lag.time.max.ms 參數(shù)設(shè)定。leader 發(fā)生故障之后，就會(huì)從 ISR 中選舉新的 leader。(之前還有另一個(gè)參數(shù)，0.9 版本之后 replica.lag.max.messages 參數(shù)被移除了)

c) ack應(yīng)答機(jī)制

對(duì)于某些不太重要的數(shù)據(jù)，對(duì)數(shù)據(jù)的可靠性要求不是很高，能夠容忍數(shù)據(jù)的少量丟失，所以沒(méi)必要等 ISR 中的follower全部接收成功。

所以Kafka為用戶提供了三種可靠性級(jí)別，用戶根據(jù)對(duì)可靠性和延遲的要求進(jìn)行權(quán)衡，選擇以下的acks 參數(shù)配置

0：producer 不等待 broker 的 ack，這一操作提供了一個(gè)最低的延遲，broker 一接收到還沒(méi)有寫(xiě)入磁盤就已經(jīng)返回，當(dāng) broker 故障時(shí)有可能丟失數(shù)據(jù);

1：producer 等待 broker 的 ack，partition 的 leader 落盤成功后返回 ack，如果在 follower 同步成功之前 leader 故障，那么將會(huì)丟失數(shù)據(jù);

-1(all)：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盤成功后才返回 ack。但是 如果在 follower 同步完成后，broker 發(fā)送 ack 之前，leader 發(fā)生故障，那么就會(huì)造成數(shù)據(jù)重復(fù)。

d) 故障處理

由于我們并不能保證 Kafka 集群中每時(shí)每刻 follower 的長(zhǎng)度都和 leader 一致(即數(shù)據(jù)同步是有時(shí)延的)，那么當(dāng)leader 掛掉選舉某個(gè) follower 為新的 leader 的時(shí)候(原先掛掉的 leader 恢復(fù)了成為了 follower)，可能會(huì)出現(xiàn)leader 的數(shù)據(jù)比 follower 還少的情況。為了解決這種數(shù)據(jù)量不一致帶來(lái)的混亂情況，Kafka 提出了以下概念：

• LEO(Log End Offset)：指的是每個(gè)副本最后一個(gè)offset;
• HW(High Wather)：指的是消費(fèi)者能見(jiàn)到的最大的 offset，ISR 隊(duì)列中最小的 LEO。

消費(fèi)者和 leader 通信時(shí)，只能消費(fèi) HW 之前的數(shù)據(jù)，HW 之后的數(shù)據(jù)對(duì)消費(fèi)者不可見(jiàn)。

針對(duì)這個(gè)規(guī)則：

• 當(dāng)follower發(fā)生故障時(shí)：follower 發(fā)生故障后會(huì)被臨時(shí)踢出 ISR，待該 follower 恢復(fù)后，follower 會(huì)讀取本地磁盤記錄的上次的 HW，并將 log 文件高于 HW 的部分截取掉，從 HW 開(kāi)始向 leader 進(jìn)行同步。等該 follower 的 LEO 大于等于該 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。
• 當(dāng)leader發(fā)生故障時(shí)：leader 發(fā)生故障之后，會(huì)從 ISR 中選出一個(gè)新的 leader，之后，為保證多個(gè)副本之間的數(shù)據(jù)一致性，其余的 follower 會(huì)先將各自的 log 文件高于 HW 的部分截掉，然后從新的 leader 同步數(shù)據(jù)。

所以數(shù)據(jù)一致性并不能保證數(shù)據(jù)不丟失或者不重復(fù)，這是由 ack 控制的。HW 規(guī)則只能保證副本之間的數(shù)據(jù)一致性!

2.5 Exactly Once語(yǔ)義

將服務(wù)器的 ACK 級(jí)別設(shè)置為 -1，可以保證 Producer 到 Server 之間不會(huì)丟失數(shù)據(jù)，即 At Least Once 語(yǔ)義。相對(duì)的，將服務(wù)器 ACK 級(jí)別設(shè)置為 0，可以保證生產(chǎn)者每條消息只會(huì)被發(fā)送一次，即 At Most Once語(yǔ)義。

At Least Once 可以保證數(shù)據(jù)不丟失，但是不能保證數(shù)據(jù)不重復(fù)。相對(duì)的，At Most Once 可以保證數(shù)據(jù)不重復(fù)，但是不能保證數(shù)據(jù)不丟失。但是，對(duì)于一些非常重要的信息，比如說(shuō)交易數(shù)據(jù)，下游數(shù)據(jù)消費(fèi)者要求數(shù)據(jù)既不重復(fù)也不丟失，即 Exactly Once 語(yǔ)義。在 0.11 版本以前的 Kafka，對(duì)此是無(wú)能為力的，只能保證數(shù)據(jù)不丟失，再在下游消費(fèi)者對(duì)數(shù)據(jù)做全局去重。對(duì)于多個(gè)下游應(yīng)用的情況，每個(gè)都需要單獨(dú)做全局去重，這就對(duì)性能造成了很大的影響。

0.11 版本的 Kafka，引入了一項(xiàng)重大特性：冪等性。所謂的冪等性就是指 Producer 不論向 Server 發(fā)送多少次重復(fù)數(shù)據(jù)。Server 端都會(huì)只持久化一條，冪等性結(jié)合 At Least Once 語(yǔ)義，就構(gòu)成了 Kafka 的 Exactily Once 語(yǔ)義，即：At Least Once + 冪等性 = Exactly Once

要啟用冪等性，只需要將 Producer 的參數(shù)中 enable.idompotence 設(shè)置為 true 即可。Kafka 的冪等性實(shí)現(xiàn)其實(shí)就是將原來(lái)下游需要做的去重放在了數(shù)據(jù)上游。開(kāi)啟冪等性的 Producer 在初始化的時(shí)候會(huì)被分配一個(gè) PID，發(fā)往同一 Partition 的消息會(huì)附帶 Sequence Number。而 Broker 端會(huì)對(duì)

但是 PID 重啟就會(huì)變化，同時(shí)不同的 Partition 也具有不同主鍵，所以冪等性無(wú)法保證跨分區(qū)會(huì)話的 Exactly Once。

三、Broker 保存消息

3.1 存儲(chǔ)方式

物理上把 topic 分成一個(gè)或多個(gè) patition(對(duì)應(yīng) server.properties 中的 num.partitions=3 配置)，每個(gè) patition 物理上對(duì)應(yīng)一個(gè)文件夾(該文件夾存儲(chǔ)該 patition 的所有消息和索引文件)。

3.2 存儲(chǔ)策略

無(wú)論消息是否被消費(fèi)， kafka 都會(huì)保留所有消息。有兩種策略可以刪除舊數(shù)據(jù)：

基于時(shí)間：log.retention.hours=168

基于大小：log.retention.bytes=1073741824 需要注意的是，因?yàn)?Kafka 讀取特定消息的時(shí)間復(fù)雜度為 O(1)，即與文件大小無(wú)關(guān)， 所以這里刪除過(guò)期文件與提高 Kafka 性能無(wú)關(guān)。

四、Kafka 消費(fèi)過(guò)程

Kafka 消費(fèi)者采用 pull 拉模式從 broker 中消費(fèi)數(shù)據(jù)。與之相對(duì)的 push(推)模式很難適應(yīng)消費(fèi)速率不同的消費(fèi)者，因?yàn)橄l(fā)送速率是由 broker 決定的。它的目標(biāo)是盡可能以最快速度傳遞消息，但是這樣很容易造成 consumer 來(lái)不及處理消息。而 pull 模式則可以根據(jù) consumer 的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息。

pull 模式不足之處是，如果 kafka 沒(méi)有數(shù)據(jù)，消費(fèi)者可能會(huì)陷入循環(huán)中，一直返回空數(shù)據(jù)。為了避免這種情況，我們?cè)谖覀兊睦?qǐng)求中有參數(shù)，允許消費(fèi)者請(qǐng)求在等待數(shù)據(jù)到達(dá)的“長(zhǎng)輪詢”中進(jìn)行阻塞(并且可選地等待到給定的字節(jié)數(shù)，以確保大的傳輸大小，或者傳入等待超時(shí)時(shí)間)。

4.1 消費(fèi)者組

消費(fèi)者是以 consumer group 消費(fèi)者組的方式工作，由一個(gè)或者多個(gè)消費(fèi)者組成一個(gè)組， 共同消費(fèi)一個(gè) topic。每個(gè)分區(qū)在同一時(shí)間只能由 group 中的一個(gè)消費(fèi)者讀取，但是多個(gè) group 可以同時(shí)消費(fèi)這個(gè) partition。在圖中，有一個(gè)由三個(gè)消費(fèi)者組成的 group，有一個(gè)消費(fèi)者讀取主題中的兩個(gè)分區(qū)，另外兩個(gè)分別讀取一個(gè)分區(qū)。某個(gè)消費(fèi)者讀取某個(gè)分區(qū)，也可以叫做某個(gè)消費(fèi)者是某個(gè)分區(qū)的擁有者。

在這種情況下，消費(fèi)者可以通過(guò)水平擴(kuò)展的方式同時(shí)讀取大量的消息。另外，如果一個(gè)消費(fèi)者失敗了，那么其他的 group 成員會(huì)自動(dòng)負(fù)載均衡讀取之前失敗的消費(fèi)者讀取的分區(qū)。

消費(fèi)者組最為重要的一個(gè)功能是實(shí)現(xiàn)廣播與單播的功能。一個(gè)消費(fèi)者組可以確保其所訂閱的 Topic 的每個(gè)分區(qū)只能被從屬于該消費(fèi)者組中的唯一一個(gè)消費(fèi)者所消費(fèi);如果不同的消費(fèi)者組訂閱了同一個(gè) Topic，那么這些消費(fèi)者組之間是彼此獨(dú)立的，不會(huì)受到相互的干擾。

如果我們希望一條消息可以被多個(gè)消費(fèi)者所消費(fèi)，那么可以將這些消費(fèi)者放到不同的消費(fèi)者組中，這實(shí)際上就是廣播的效果;如果希望一條消息只能被一個(gè)消費(fèi)者所消費(fèi)，那么可以將這些消費(fèi)者放到同一個(gè)消費(fèi)者組中，這實(shí)際上就是單播的效果。

4.2 分區(qū)分配策略

一個(gè) consumer group 中有多個(gè) consumer，一個(gè) topic 有多個(gè) partition，所以必然會(huì)涉及到 partition 的分配問(wèn)題，即確定哪個(gè) partition 由哪個(gè) consumer 來(lái)消費(fèi)。

Kafka 有兩種分配策略，一是 RoundRobin，一是 Range。

RoundRobin

RoundRobin 即輪詢的意思，比如現(xiàn)在有一個(gè)三個(gè)消費(fèi)者 ConsumerA、ConsumerB 和 ConsumerC 組成的消費(fèi)者組，同時(shí)消費(fèi) TopicA 主題消息，TopicA 分為 7 個(gè)分區(qū)，如果采用 RoundRobin 分配策略，過(guò)程如下所示：

圖片：mrbird.cc

這種輪詢的方式應(yīng)該很好理解。但如果消費(fèi)者組消費(fèi)多個(gè)主題的多個(gè)分區(qū)，會(huì)發(fā)生什么情況呢?比如現(xiàn)在有一個(gè)兩個(gè)消費(fèi)者 ConsumerA 和 ConsumerB 組成的消費(fèi)者組，同時(shí)消費(fèi) TopicA 和 TopicB 主題消息，如果采用RoundRobin 分配策略，過(guò)程如下所示：

注：TAP0 表示 TopicA Partition0 分區(qū)數(shù)據(jù)，以此類推。

這種情況下，采用 RoundRobin 算法分配，多個(gè)主題會(huì)被當(dāng)做一個(gè)整體來(lái)看，這個(gè)整體包含了各自的 Partition，比如在 Kafka-clients 依賴中，與之對(duì)應(yīng)的對(duì)象為 TopicPartition。接著將這些 TopicPartition 根據(jù)其哈希值進(jìn)行排序，排序后采用輪詢的方式分配給消費(fèi)者。

但這會(huì)帶來(lái)一個(gè)問(wèn)題：假如上圖中的消費(fèi)者組中，ConsumerA 只訂閱了 TopicA 主題，ConsumerB 只訂閱了TopicB 主題，采用 RoundRobin 輪詢算法后，可能會(huì)出現(xiàn) ConsumerA 消費(fèi)了 TopicB 主題分區(qū)里的消息，ConsumerB 消費(fèi)了 TopicA 主題分區(qū)里的消息。

綜上所述，RoundRobin 算法只適用于消費(fèi)者組中消費(fèi)者訂閱的主題相同的情況。同時(shí)會(huì)發(fā)現(xiàn)，采用 RoundRobin 算法，消費(fèi)者組里的消費(fèi)者之間消費(fèi)的消息個(gè)數(shù)最多相差 1 個(gè)。

Range

Kafka 默認(rèn)采用 Range 分配策略，Range 顧名思義就是按范圍劃分的意思。

比如現(xiàn)在有一個(gè)三個(gè)消費(fèi)者 ConsumerA、ConsumerB 和 ConsumerC 組成的消費(fèi)者組，同時(shí)消費(fèi) TopicA 主題消息，TopicA分為7個(gè)分區(qū)，如果采用 Range 分配策略，過(guò)程如下所示：

假如現(xiàn)在有一個(gè)兩個(gè)消費(fèi)者 ConsumerA 和 ConsumerB 組成的消費(fèi)者組，同時(shí)消費(fèi) TopicA 和 TopicB 主題消息，如果采用 Range 分配策略，過(guò)程如下所示：

Range 算法并不會(huì)把多個(gè)主題分區(qū)當(dāng)成一個(gè)整體。

從上面的例子我們可以總結(jié)出Range算法的一個(gè)弊端：那就是同一個(gè)消費(fèi)者組內(nèi)的消費(fèi)者消費(fèi)的消息數(shù)量相差可能較大。

4.3 offset 的維護(hù)

由于 consumer 在消費(fèi)過(guò)程中可能會(huì)出現(xiàn)斷電宕機(jī)等故障，consumer 恢復(fù)后，需要從故障前的位置繼續(xù)消費(fèi)，所以 consumer 需要實(shí)時(shí)記錄自己消費(fèi)到了哪個(gè) offset，以便故障恢復(fù)后繼續(xù)消費(fèi)。

Kafka 0.9 版本之前，consumer 默認(rèn)將 offset 保存在 Zookeeper 中，從 0.9 版本開(kāi)始，consumer 默認(rèn)將 offset保存在 Kafka 一個(gè)內(nèi)置的 topic 中，該 topic 為 _consumer_offsets。

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic starfish --from-beginning 
one 

消費(fèi) topic 后，查看 kafka-logs 目錄，會(huì)發(fā)現(xiàn)多出 50 個(gè)分區(qū)。

默認(rèn)情況下__consumer_offsets 有 50 個(gè)分區(qū)，如果你的系統(tǒng)中 consumer group 也很多的話，那么這個(gè)命令的輸出結(jié)果會(huì)很多

五、Kafka事務(wù)

Kafka 從 0.11 版本開(kāi)始引入了事務(wù)支持。事務(wù)可以保證 Kafka 在 Exactly Once 語(yǔ)義的基礎(chǔ)上，生產(chǎn)和消費(fèi)可以跨分區(qū)和會(huì)話，要么全部成功，要么全部失敗。

5.1 Producer事務(wù)

為了了實(shí)現(xiàn)跨分區(qū)跨會(huì)話的事務(wù)，需要引入一個(gè)全局唯一的 TransactionID，并將 Producer 獲得的 PID 和Transaction ID 綁定。這樣當(dāng) Producer 重啟后就可以通過(guò)正在進(jìn)行的 TransactionID 獲得原來(lái)的 PID。

為了管理 Transaction，Kafka 引入了一個(gè)新的組件 Transaction Coordinator。Producer 就是通過(guò)和 Transaction Coordinator 交互獲得 Transaction ID 對(duì)應(yīng)的任務(wù)狀態(tài)。Transaction Coordinator 還負(fù)責(zé)將事務(wù)所有寫(xiě)入 Kafka 的一個(gè)內(nèi)部 Topic，這樣即使整個(gè)服務(wù)重啟，由于事務(wù)狀態(tài)得到保存，進(jìn)行中的事務(wù)狀態(tài)可以得到恢復(fù)，從而繼續(xù)進(jìn)行。

5.2 Consumer事務(wù)

對(duì) Consumer 而言，事務(wù)的保證就會(huì)相對(duì)較弱，尤其是無(wú)法保證 Commit 的消息被準(zhǔn)確消費(fèi)。這是由于Consumer 可以通過(guò) offset 訪問(wèn)任意信息，而且不同的 SegmentFile 生命周期不同，同一事務(wù)的消息可能會(huì)出現(xiàn)重啟后被刪除的情況。

參考：

尚硅谷Kafka教學(xué)

部分圖片來(lái)源：mrbird.cc

https://gitbook.cn/books/5ae1e77197c22f130e67ec4e/index.html