場景描述：

Kafka使用分區(qū)將topic的消息打散到多個分區(qū)分布保存在不同的broker上，實現(xiàn)了producer和consumer消息處理的高吞吐量。
Kafka的producer和consumer都可以多線程地并行操作，而每個線程處理的是一個分區(qū)的數(shù)據(jù)。因此分區(qū)實際上是調(diào)優(yōu)Kafka并行度的最小單元。
對于producer而言，它實際上是用多個線程并發(fā)地向不同分區(qū)所在的broker發(fā)起Socket連接同時給這些分區(qū)發(fā)送消息；而consumer，同一個消費組內(nèi)的所有consumer線程都被指定topic的某一個分區(qū)進行消費。

所以說，如果一個topic分區(qū)越多，理論上整個集群所能達到的吞吐量就越大。

分區(qū)多的優(yōu)點

一個topic分區(qū)越多，理論上整個集群所能達到的吞吐量就越大
理由:
kafka使用分區(qū)將topic的消息打散到多個分區(qū)分布保存在不同的broker上，實現(xiàn)了producer和consumer消息處理的高吞吐量。
Kafka的producer和consumer都可以多線程地并行操作，而每個線程處理的是一個分區(qū)的數(shù)據(jù)。因此分區(qū)實際上是調(diào)優(yōu)Kafka并行度的最小單元。
對于producer而言，它實際上是用多個線程并發(fā)地向不同分區(qū)所在的broker發(fā)起Socket連接同時給這些分區(qū)發(fā)送消息；
而consumer，同一個消費組內(nèi)的所有consumer線程都被指定topic的某一個分區(qū)進行消費。

分區(qū)不是越多越好

每個分區(qū)都有自己的開銷：

一、客戶端/服務(wù)器端需要使用的內(nèi)存就越多

• Kafka0.8.2之后，在客戶端producer有個參數(shù)batch.size，默認是16KB。
  它會為每個分區(qū)緩存消息，一旦滿了就打包將消息批量發(fā)出?？瓷先ミ@是個能夠提升性能的設(shè)計。不過很顯然，因為這個參數(shù)是分區(qū)級別的，如果分區(qū)數(shù)越多，這部分緩存所需的內(nèi)存占用也會更多。
  假設(shè)你有10000個分區(qū)，按照默認設(shè)置，這部分緩存需要占用約157MB的內(nèi)存。而consumer端呢？我們拋開獲取數(shù)據(jù)所需的內(nèi)存不說，只說線程的開銷。如果還是假設(shè)有10000個分區(qū)，同時consumer線程數(shù)要匹配分區(qū)數(shù)(大部分情況下是最佳的消費吞吐量配置)的話，那么在consumer client就要創(chuàng)建10000個線程，也需要創(chuàng)建大約10000個Socket去獲取分區(qū)數(shù)據(jù)。這里面的線程切換的開銷本身已經(jīng)不容小覷了。

• 服務(wù)器端的開銷也不小，如果閱讀Kafka源碼的話可以發(fā)現(xiàn)，服務(wù)器端的很多組件都在內(nèi)存中維護了分區(qū)級別的緩存，比如controller，F(xiàn)etcherManager等，因此分區(qū)數(shù)越多，這種緩存的成本就越大。

二、文件句柄的開銷

每個分區(qū)在底層文件系統(tǒng)都有屬于自己的一個目錄。
該目錄下通常會有兩個文件：base_offset.log和base_offset.index。
Kafak的controller和ReplicaManager會為每個broker都保存這兩個文件句柄(file handler)。很明顯，如果分區(qū)數(shù)越多，所需要保持打開狀態(tài)的文件句柄數(shù)也就越多，最終可能會突破你的ulimit -n的限制。

三、降低高可用性

Kafka通過副本(replica)機制來保證高可用。
具體做法就是為每個分區(qū)保存若干個副本(replica_factor指定副本數(shù))。每個副本保存在不同的broker上。
其中的一個副本充當leader 副本，負責處理producer和consumer請求。其他副本充當follower角色，由Kafka controller負責保證與leader的同步。
如果leader所在的broker掛掉了，contorller會檢測到,然后在zookeeper的幫助下重選出新的leader——這中間會有短暫的不可用時間窗口，雖然大部分情況下可能只是幾毫秒級別。但如果你有10000個分區(qū)，10個broker，也就是說平均每個broker上有1000個分區(qū)。
此時這個broker掛掉了，那么zookeeper和controller需要立即對這1000個分區(qū)進行l(wèi)eader選舉。
比起很少的分區(qū)leader選舉而言，這必然要花更長的時間，并且通常不是線性累加的。如果這個broker還同時是controller情況就更糟了。

如何確定分區(qū)數(shù)量呢？

可以遵循一定的步驟來嘗試確定分區(qū)數(shù)：創(chuàng)建一個只有1個分區(qū)的topic，然后測試這個topic的producer吞吐量和consumer吞吐量。假設(shè)它們的值分別是Tp和Tc，單位可以是MB/s。然后假設(shè)總的目標吞吐量是Tt，那么分區(qū)數(shù) = Tt / max(Tp, Tc)

說明：Tp表示producer的吞吐量。測試producer通常是很容易的，因為它的邏輯非常簡單，就是直接發(fā)送消息到Kafka就好了。Tc表示consumer的吞吐量。測試Tc通常與應(yīng)用的關(guān)系更大， 因為Tc的值取決于你拿到消息之后執(zhí)行什么操作，因此Tc的測試通常也要麻煩一些。

一條消息如何知道要被發(fā)送到哪個分區(qū)？

按照key值分配

默認情況下，Kafka根據(jù)傳遞消息的key來進行分區(qū)的分配，即hash(key) % numPartitions：

def partition(key: Any, numPartitions: Int): Int = { Utils.abs(key.hashCode) % numPartitions }

這保證了相同key的消息一定會被路由到相同的分區(qū)。key為null時，從緩存中取分區(qū)id或者隨機取一個。如果你沒有指定key，那么Kafka是如何確定這條消息去往哪個分區(qū)的呢？

圖片

不指定key時，Kafka幾乎就是隨機找一個分區(qū)發(fā)送無key的消息，然后把這個分區(qū)號加入到緩存中以備后面直接使用——當然了，Kafka本身也會清空該緩存（默認每10分鐘或每次請求topic元數(shù)據(jù)時）。

Consumer個數(shù)與分區(qū)數(shù)有什么關(guān)系？

topic下的一個分區(qū)只能被同一個consumer group下的一個consumer線程來消費，但反之并不成立，即一個consumer線程可以消費多個分區(qū)的數(shù)據(jù)，比如Kafka提供的ConsoleConsumer，默認就只是一個線程來消費所有分區(qū)的數(shù)據(jù)。

圖片

所以，如果你的分區(qū)數(shù)是N，那么最好線程數(shù)也保持為N，這樣通常能夠達到最大的吞吐量。超過N的配置只是浪費系統(tǒng)資源，因為多出的線程不會被分配到任何分區(qū)。

Consumer消費Partition的分配策略

Kafka提供的兩種分配策略：range和roundrobin，由參數(shù)partition.assignment.strategy指定，默認是range策略。

當以下事件發(fā)生時，Kafka 將會進行一次分區(qū)分配：

• 同一個 Consumer Group 內(nèi)新增消費者

• 消費者離開當前所屬的Consumer Group，包括shuts down 或 crashes

• 訂閱的主題新增分區(qū)

將分區(qū)的所有權(quán)從一個消費者移到另一個消費者稱為重新平衡（rebalance），如何rebalance就涉及到本文提到的分區(qū)分配策略。

下面我們將詳細介紹 Kafka 內(nèi)置的兩種分區(qū)分配策略。本文假設(shè)我們有個名為 T1 的主題，其包含了10個分區(qū)，然后我們有兩個消費者（C1，C2）

來消費這10個分區(qū)里面的數(shù)據(jù)，而且 C1 的 num.streams = 1，C2 的 num.streams = 2。

Range strategy

Range策略是對每個主題而言的，首先對同一個主題里面的分區(qū)按照序號進行排序，并對消費者按照字母順序進行排序。在我們的例子里面，排完序的分區(qū)將會是0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消費者線程排完序?qū)荂1-0, C2-0, C2-1。然后將partitions的個數(shù)除于消費者線程的總數(shù)來決定每個消費者線程消費幾個分區(qū)。如果除不盡，那么前面幾個消費者線程將會多消費一個分區(qū)。在我們的例子里面，我們有10個分區(qū)，3個消費者線程， 10 / 3 = 3，而且除不盡，那么消費者線程 C1-0 將會多消費一個分區(qū)，所以最后分區(qū)分配的結(jié)果看起來是這樣的：

• C1-0 將消費 0, 1, 2, 3 分區(qū)
• C2-0 將消費 4, 5, 6 分區(qū)
• C2-1 將消費 7, 8, 9 分區(qū)

假如我們有11個分區(qū)，那么最后分區(qū)分配的結(jié)果看起來是這樣的：

• C1-0 將消費 0, 1, 2, 3 分區(qū)
• C2-0 將消費 4, 5, 6, 7 分區(qū)
• C2-1 將消費 8, 9, 10 分區(qū)

假如我們有2個主題(T1和T2)，分別有10個分區(qū)，那么最后分區(qū)分配的結(jié)果看起來是這樣的：

• C1-0 將消費 T1主題的 0, 1, 2, 3 分區(qū)以及 T2主題的 0, 1, 2, 3分區(qū)
• C2-0 將消費 T1主題的 4, 5, 6 分區(qū)以及 T2主題的 4, 5, 6分區(qū)
• C2-1 將消費 T1主題的 7, 8, 9 分區(qū)以及 T2主題的 7, 8, 9分區(qū)

可以看出，C1-0 消費者線程比其他消費者線程多消費了2個分區(qū)，這就是Range strategy的一個很明顯的弊端。

RoundRobin strategy

使用RoundRobin策略有兩個前提條件必須滿足：
同一個Consumer Group里面的所有消費者的num.streams必須相等；
每個消費者訂閱的主題必須相同。

所以這里假設(shè)前面提到的2個消費者的num.streams = 2。RoundRobin策略的工作原理：將所有主題的分區(qū)組成 TopicAndPartition 列表，然后對 TopicAndPartition 列表按照 hashCode 進行排序，看下面的代碼應(yīng)該會明白：

圖片

最后按照round-robin風格將分區(qū)分別分配給不同的消費者線程。

在這個的例子里面，假如按照 hashCode 排序完的topic-partitions組依次為T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我們的消費者線程排序為C1-0, C1-1, C2-0, C2-1，最后分區(qū)分配的結(jié)果為：

• C1-0 將消費 T1-5, T1-2, T1-6 分區(qū)；
• C1-1 將消費 T1-3, T1-1, T1-9 分區(qū)；
• C2-0 將消費 T1-0, T1-4 分區(qū)；
• C2-1 將消費 T1-8, T1-7 分區(qū)；

多個主題的分區(qū)分配和單個主題類似。遺憾的是，目前我們還不能自定義分區(qū)分配策略，只能通過partition.assignment.strategy參數(shù)選擇 range 或 roundrobin。

Kafka的每個分區(qū)只能被一個線程消費，如何做到多個線程同時消費一個分區(qū)?

假設(shè)多個消費者負責同一個分區(qū)，那么會有什么問題呢？

我們知道，Kafka它在設(shè)計的時候就是要保證分區(qū)下消息的順序，也就是說消息在一個分區(qū)中的順序是怎樣的，那么消費者在消費的時候看到的就是什么樣的順序，那么要做到這一點就首先要保證消息是由消費者主動拉取的（pull），其次還要保證一個分區(qū)只能由一個消費者負責。倘若，兩個消費者負責同一個分區(qū)，那么就意味著兩個消費者同時讀取分區(qū)的消息，由于消費者自己可以控制讀取消息的offset，就有可能C1才讀到2，而C1讀到1，C1還沒處理完，C2已經(jīng)讀到3了，則會造成很多浪費，因為這就相當于多線程讀取同一個消息，會造成消息處理的重復，且不能保證消息的順序，這就跟主動推送（push）無異。

實現(xiàn)多線程步驟如下：

生產(chǎn)者隨機分區(qū)提交數(shù)據(jù)(自定義隨機分區(qū))。
消費者修改單線程模式為多線程，在消費方面得注意，得遍歷所有分區(qū)，否則還是只消費了一個區(qū)。

Kafka Producer是如何動態(tài)感知Topic分區(qū)數(shù)變化的?

在啟動Kafka Producer往Kafka的Broker發(fā)送消息的時候，用戶修改了該Topic的分區(qū)數(shù)，Producer可以在最多topic.metadata.refresh.interval.ms的時間之后感知到，并且可以將數(shù)據(jù)發(fā)送到新添加的分區(qū)中。

Kafka分區(qū)數(shù)可以增加或減少嗎?為什么?

支持增加分區(qū),不支持 減少分區(qū)

我們可以使用 bin/kafka-topics.sh 命令對 Kafka 增加 分區(qū) ，但是 Kafka 不支持減少分區(qū)數(shù)。
Kafka 分區(qū)數(shù)據(jù)不支持減少是由很多原因的，比如減少的分區(qū)其數(shù)據(jù)放到哪里去？
是刪除，還是保留？
刪除的話，那么這些沒消費的消息不就丟了。
如果保留這些消息如何放到其他分區(qū)里面？
追加到其他分區(qū)后面的話那么就破壞了 Kafka 單個分區(qū)的有序性。
如果要保證刪除分區(qū)數(shù)據(jù)插入到其他分區(qū)保證有序性，那么實現(xiàn)起來邏輯就會非常復雜。