問(wèn)題

用過(guò) Kafka 的同學(xué)用過(guò)都知道，每個(gè) Topic 一般會(huì)有很多個(gè) partitions。為了使得我們能夠及時(shí)消費(fèi)消息，我們也可能會(huì)啟動(dòng)多個(gè) Consumer 去消費(fèi)，而每個(gè) Consumer 又會(huì)啟動(dòng)一個(gè)或多個(gè)streams去分別消費(fèi) Topic 里面的數(shù)據(jù)。我們又知道，Kafka 存在 Consumer Group 的概念，也就是 group.id 一樣的 Consumer，這些 Consumer 屬于同一個(gè)Consumer Group，組內(nèi)的所有消費(fèi)者協(xié)調(diào)在一起來(lái)消費(fèi)訂閱主題(subscribed topics)的所有分區(qū)(partition)。當(dāng)然，每個(gè)分區(qū)只能由同一個(gè)消費(fèi)組內(nèi)的一個(gè)consumer來(lái)消費(fèi)。那么問(wèn)題來(lái)了，同一個(gè) Consumer Group 里面的 Consumer 是如何知道該消費(fèi)哪些分區(qū)里面的數(shù)據(jù)呢？

如上圖，Consumer1 為啥消費(fèi)的是 Partition0 和 Partition2，而不是 Partition0 和 Partition3？這就涉及到 Kafka 內(nèi)部分區(qū)分配策略（Partition Assignment Strategy）了。

在 Kafka 內(nèi)部存在兩種默認(rèn)的分區(qū)分配策略：Range 和 RoundRobin。當(dāng)以下事件發(fā)生時(shí)，Kafka 將會(huì)進(jìn)行一次分區(qū)分配：
?同一個(gè) Consumer Group 內(nèi)新增消費(fèi)者
?消費(fèi)者離開(kāi)當(dāng)前所屬的Consumer Group，包括shuts down 或 crashes
?訂閱的主題新增分區(qū)

將分區(qū)的所有權(quán)從一個(gè)消費(fèi)者移到另一個(gè)消費(fèi)者稱為重新平衡（rebalance），如何rebalance就涉及到本文提到的分區(qū)分配策略。下面我們將詳細(xì)介紹 Kafka 內(nèi)置的兩種分區(qū)分配策略。本文假設(shè)我們有個(gè)名為 T1 的主題，其包含了10個(gè)分區(qū)，然后我們有兩個(gè)消費(fèi)者（C1，C2）來(lái)消費(fèi)這10個(gè)分區(qū)里面的數(shù)據(jù)，而且 C1 的 num.streams = 1，C2 的 num.streams = 2。

Range strategy

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

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

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

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

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

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

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

RoundRobin strategy

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

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

val allTopicPartitions = ctx.partitionsForTopic.flatMap { case(topic, partitions) =>info("Consumer %s rebalancing the following partitions for topic %s: %s".format(ctx.consumerId, topic, partitions))partitions.map(partition => {TopicAndPartition(topic, partition)}) }.toSeq.sortWith((topicPartition1, topicPartition2) => {/** Randomize the order by taking the hashcode to reduce the likelihood of all partitions of a given topic ending* up on one consumer (if it has a high enough stream count).*/topicPartition1.toString.hashCode < topicPartition2.toString.hashCode })

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

在我們的例子里面，加入按照 hashCode 排序完的topic-partitions組依次為T(mén)1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我們的消費(fèi)者線程排序?yàn)镃1-0, C1-1, C2-0, C2-1，最后分區(qū)分配的結(jié)果為：

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

多個(gè)主題的分區(qū)分配和單個(gè)主題類(lèi)似，這里就不在介紹了。

根據(jù)上面的詳細(xì)介紹相信大家已經(jīng)對(duì)Kafka的分區(qū)分配策略原理很清楚了。不過(guò)遺憾的是，目前我們還不能自定義分區(qū)分配策略，只能通過(guò)partition.assignment.strategy參數(shù)選擇 range 或 roundrobin。partition.assignment.strategy參數(shù)默認(rèn)的值是range。

總結(jié)

以上是生活随笔為你收集整理的Kafka分区分配策略(Partition Assignment Strategy)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。