問題

用過 Kafka 的同學用過都知道，每個 Topic 一般會有很多個 partitions。為了使得我們能夠及時消費訊息，我們也可能會啟動多個 Consumer 去消費，而每個 Consumer 又會啟動一個或多個streams去分別消費 Topic 裡面的資料。我們又知道，Kafka 存在 Consumer Group 的概念，也就是 group.id 一樣的 Consumer，這些 Consumer 屬於同一個Consumer Group，組內的所有消費者協調在一起來消費訂閱主題(subscribed topics)的所有分割槽(partition)。當然，每個分割槽只能由同一個消費組內的一個consumer來消費。那麼問題來了，同一個 Consumer Group 裡面的 Consumer 是如何知道該消費哪些分割槽裡面的資料呢？

在 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。