Kafka分区分配策略(Partition Assignment Strategy)
問題
用過 Kafka 的同學用過都知道,每個 Topic 一般會有很多個 partitions。為了使得我們能夠及時消費消息,我們也可能會啟動多個 Consumer 去消費,而每個 Consumer 又會啟動一個或多個streams去分別消費 Topic 里面的數據。我們又知道,Kafka 存在 Consumer Group 的概念,也就是 group.id 一樣的 Consumer,這些 Consumer 屬于同一個Consumer Group,組內的所有消費者協調在一起來消費訂閱主題(subscribed topics)的所有分區(partition)。當然,每個分區只能由同一個消費組內的一個consumer來消費。那么問題來了,同一個 Consumer Group 里面的 Consumer 是如何知道該消費哪些分區里面的數據呢?
如上圖,Consumer1 為啥消費的是 Partition0 和 Partition2,而不是 Partition0 和 Partition3?這就涉及到 Kafka 內部分區分配策略(Partition Assignment Strategy)了。
在 Kafka 內部存在兩種默認的分區分配策略:Range 和 RoundRobin。當以下事件發生時,Kafka 將會進行一次分區分配:
?同一個 Consumer Group 內新增消費者
?消費者離開當前所屬的Consumer Group,包括shuts down 或 crashes
?訂閱的主題新增分區
將分區的所有權從一個消費者移到另一個消費者稱為重新平衡(rebalance),如何rebalance就涉及到本文提到的分區分配策略。下面我們將詳細介紹 Kafka 內置的兩種分區分配策略。本文假設我們有個名為 T1 的主題,其包含了10個分區,然后我們有兩個消費者(C1,C2)來消費這10個分區里面的數據,而且 C1 的 num.streams = 1,C2 的 num.streams = 2。
Range strategy
Range策略是對每個主題而言的,首先對同一個主題里面的分區按照序號進行排序,并對消費者按照字母順序進行排序。在我們的例子里面,排完序的分區將會是0, 1, 2, 3, 4, 5, 6, 7, 8, 9;消費者線程排完序將會是C1-0, C2-0, C2-1。然后將partitions的個數除于消費者線程的總數來決定每個消費者線程消費幾個分區。如果除不盡,那么前面幾個消費者線程將會多消費一個分區。在我們的例子里面,我們有10個分區,3個消費者線程, 10 / 3 = 3,而且除不盡,那么消費者線程 C1-0 將會多消費一個分區,所以最后分區分配的結果看起來是這樣的:
C1-0 將消費 0, 1, 2, 3 分區 C2-0 將消費 4, 5, 6 分區 C2-1 將消費 7, 8, 9 分區假如我們有11個分區,那么最后分區分配的結果看起來是這樣的:
C1-0 將消費 0, 1, 2, 3 分區 C2-0 將消費 4, 5, 6, 7 分區 C2-1 將消費 8, 9, 10 分區假如我們有2個主題(T1和T2),分別有10個分區,那么最后分區分配的結果看起來是這樣的:
C1-0 將消費 T1主題的 0, 1, 2, 3 分區以及 T2主題的 0, 1, 2, 3分區 C2-0 將消費 T1主題的 4, 5, 6 分區以及 T2主題的 4, 5, 6分區 C2-1 將消費 T1主題的 7, 8, 9 分區以及 T2主題的 7, 8, 9分區可以看出,C1-0 消費者線程比其他消費者線程多消費了2個分區,這就是Range strategy的一個很明顯的弊端。
RoundRobin strategy
使用RoundRobin策略有兩個前提條件必須滿足:
?同一個Consumer Group里面的所有消費者的num.streams必須相等;
?每個消費者訂閱的主題必須相同。
所以這里假設前面提到的2個消費者的num.streams = 2。RoundRobin策略的工作原理:將所有主題的分區組成 TopicAndPartition 列表,然后對 TopicAndPartition 列表按照 hashCode 進行排序,這里文字可能說不清,看下面的代碼應該會明白:
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風格將分區分別分配給不同的消費者線程。
在我們的例子里面,加入按照 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,最后分區分配的結果為:
C1-0 將消費 T1-5, T1-2, T1-6 分區; C1-1 將消費 T1-3, T1-1, T1-9 分區; C2-0 將消費 T1-0, T1-4 分區; C2-1 將消費 T1-8, T1-7 分區;多個主題的分區分配和單個主題類似,這里就不在介紹了。
根據上面的詳細介紹相信大家已經對Kafka的分區分配策略原理很清楚了。不過遺憾的是,目前我們還不能自定義分區分配策略,只能通過partition.assignment.strategy參數選擇 range 或 roundrobin。partition.assignment.strategy參數默認的值是range。
總結
以上是生活随笔為你收集整理的Kafka分区分配策略(Partition Assignment Strategy)的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: Kafka 是如何保证数据可靠性和一致性
- 下一篇: 速度与激情10谁开核潜艇?