在Kafak中國社區的qq群中，這個問題被提及的比例是相當高的，這也是Kafka用戶最常碰到的問題之一。本文結合Kafka源碼試圖對該問題相關的因素進行探討。希望對大家有所幫助。

怎么確定分區數？

“我應該選擇幾個分區？”——如果你在Kafka中國社區的群里，這樣的問題你會經常碰到的。不過有些遺憾的是，我們似乎并沒有很權威的答案能夠解答這樣的問題。其實這也不奇怪，畢竟這樣的問題通常都是沒有固定答案的。Kafka官網上標榜自己是"high-throughput distributed messaging system"，即一個高吞吐量的分布式消息引擎。那么怎么達到高吞吐量呢？Kafka在底層摒棄了Java堆緩存機制，采用了操作系統級別的頁緩存，同時將隨機寫操作改為順序寫，再結合Zero-Copy的特性極大地改善了IO性能。但是，這只是一個方面，畢竟單機優化的能力是有上限的。如何通過水平擴展甚至是線性擴展來進一步提升吞吐量呢？ Kafka就是使用了分區(partition)，通過將topic的消息打散到多個分區并分布保存在不同的broker上實現了消息處理(不管是producer還是consumer)的高吞吐量。

Kafka的生產者和消費者都可以多線程地并行操作，而每個線程處理的是一個分區的數據。因此分區實際上是調優Kafka并行度的最小單元。對于producer而言，它實際上是用多個線程并發地向不同分區所在的broker發起Socket連接同時給這些分區發送消息；而consumer呢，同一個消費組內的所有consumer線程都被指定topic的某一個分區進行消費(具體如何確定consumer線程數目我們后面會詳細說明)。所以說，如果一個topic分區越多，理論上整個集群所能達到的吞吐量就越大。

但分區是否越多越好呢？顯然也不是，因為每個分區都有自己的開銷：

一、客戶端/服務器端需要使用的內存就越多

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

服務器端的開銷也不小，如果閱讀Kafka源碼的話可以發現，服務器端的很多組件都在內存中維護了分區級別的緩存，比如controller，FetcherManager等，因此分區數越多，這種緩存的成本越久越大。

二、文件句柄的開銷

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

三、降低高可用性

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

說了這么多“廢話”，很多人肯定已經不耐煩了。那你說到底要怎么確定分區數呢？答案就是：視情況而定。基本上你還是需要通過一系列實驗和測試來確定。當然測試的依據應該是吞吐量。雖然LinkedIn這篇文章做了Kafka的基準測試，但它的結果其實對你意義不大，因為不同的硬件、軟件、負載情況測試出來的結果必然不一樣。我經常碰到的問題類似于，官網說每秒能到10MB，為什么我的producer每秒才1MB？ —— 且不說硬件條件，最后發現他使用的消息體有1KB，而官網的基準測試是用100B測出來的，因此根本沒有可比性。不過你依然可以遵循一定的步驟來嘗試確定分區數：創建一個只有1個分區的topic，然后測試這個topic的producer吞吐量和consumer吞吐量。假設它們的值分別是Tp和Tc，單位可以是MB/s。然后假設總的目標吞吐量是Tt，那么分區數 = Tt / max(Tp, Tc)

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

另外，Kafka并不能真正地做到線性擴展(其實任何系統都不能)，所以你在規劃你的分區數的時候最好多規劃一下，這樣未來擴展時候也更加方便。

消息-分區的分配

默認情況下，Kafka根據傳遞消息的key來進行分區的分配，即hash(key) % numPartitions，如下圖所示：

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

這就保證了相同key的消息一定會被路由到相同的分區。如果你沒有指定key，那么Kafka是如何確定這條消息去往哪個分區的呢？

if(key == null) { // 如果沒有指定keyval id = sendPartitionPerTopicCache.get(topic) // 先看看Kafka有沒有緩存的現成的分區Idid match {case Some(partitionId) => partitionId // 如果有的話直接使用這個分區Id就好了case None => // 如果沒有的話，val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) //找出所有可用分區的leader所在的brokerif (availablePartitions.isEmpty)throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)val index = Utils.abs(Random.nextInt) % availablePartitions.size // 從中隨機挑一個val partitionId = availablePartitions(index).partitionIdsendPartitionPerTopicCache.put(topic, partitionId) // 更新緩存以備下一次直接使用partitionId}}

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

如何設定consumer線程數

我個人的觀點，如果你的分區數是N，那么最好線程數也保持為N，這樣通常能夠達到最大的吞吐量。超過N的配置只是浪費系統資源，因為多出的線程不會被分配到任何分區。讓我們來看看具體Kafka是如何分配的。

topic下的一個分區只能被同一個consumer group下的一個consumer線程來消費，但反之并不成立，即一個consumer線程可以消費多個分區的數據，比如Kafka提供的ConsoleConsumer，默認就只是一個線程來消費所有分區的數據。——其實ConsoleConsumer可以使用通配符的功能實現同時消費多個topic數據，但這和本文無關。

再討論分配策略之前，先說說KafkaStream——它是consumer的關鍵類，提供了遍歷方法用于consumer程序調用實現數據的消費。其底層維護了一個阻塞隊列，所以在沒有新消息到來時，consumer是處于阻塞狀態的，表現出來的狀態就是consumer程序一直在等待新消息的到來。——你當然可以配置成帶超時的consumer，具體參看參數consumer.timeout.ms的用法。

下面說說 Kafka提供的兩種分配策略： range和roundrobin，由參數partition.assignment.strategy指定，默認是range策略。本文只討論range策略。所謂的range其實就是按照階段平均分配。舉個例子就明白了，假設你有10個分區，P0 ~ P9，consumer線程數是3， C0 ~ C2，那么每個線程都分配哪些分區呢？

C0 消費分區 0, 1, 2, 3

C1 消費分區 4, 5, 6

C2 消費分區 7, 8, 9

具體算法就是：

val nPartsPerConsumer = curPartitions.size / curConsumers.size // 每個consumer至少保證消費的分區數 val nConsumersWithExtraPart = curPartitions.size % curConsumers.size // 還剩下多少個分區需要單獨分配給開頭的線程們 ... for (consumerThreadId <- consumerThreadIdSet) { // 對于每一個consumer線程val myConsumerPosition = curConsumers.indexOf(consumerThreadId) //算出該線程在所有線程中的位置，介于[0, n-1]assert(myConsumerPosition >= 0) // startPart 就是這個線程要消費的起始分區數val startPart = nPartsPerConsumer * myConsumerPosition + myConsumerPosition.min(nConsumersWithExtraPart) // nParts 就是這個線程總共要消費多少個分區val nParts = nPartsPerConsumer + (if (myConsumerPosition + 1 > nConsumersWithExtraPart) 0 else 1) ... }

針對于這個例子，nPartsPerConsumer就是10/3=3，nConsumersWithExtraPart為10%3=1，說明每個線程至少保證3個分區，還剩下1個分區需要單獨分配給開頭的若干個線程。這就是為什么C0消費4個分區，后面的2個線程每個消費3個分區，具體過程詳見下面的Debug截圖信息：
ctx.myTopicThreadIds

nPartsPerConsumer = 10 / 3 = 3

nConsumersWithExtraPart = 10 % 3 = 1

第一次：

myConsumerPosition = 1

startPart = 1 * 3 + min(1, 1) = 4 ---也就是從分區4開始讀

nParts = 3 + (if (1 + 1 > 1) 0 else 1) = 3 讀取3個分區， 即4,5,6

第二次：

myConsumerPosition = 0

startPart = 3 * 0 + min(1, 0) =0 --- 從分區0開始讀

nParts = 3 + (if (0 + 1 > 1) 0 else 1) = 4 讀取4個分區，即0,1,2,3

第三次：

myConsumerPosition = 2

startPart = 3 * 2 + min(2, 1) = 7 --- 從分區7開始讀

nParts = 3 + if (2 + 1 > 1) 0 else 1) = 3 讀取3個分區，即7, 8, 9

至此10個分區都已經分配完畢

說到這里，經常有個需求就是我想讓某個consumer線程消費指定的分區而不消費其他的分區。坦率來說，目前Kafka并沒有提供自定義分配策略。做到這點很難，但仔細想一想，也許我們期望Kafka做的事情太多了，畢竟它只是個消息引擎，在Kafka中加入消息消費的邏輯也許并不是Kafka該做的事情。

不消費問題

第一步：參看消費者的基本情況
查看mwbops系統，【Consumer監控】-->【對應的consumerId】

如果offset數字一直在動，說明一直在消費，說明不存在問題，return;

如果offset數字一直不動，看Owner是不是有值存在

如果Owner是空，說明消費端的程序已經跟Kafka斷開連接，應該排查消費端是否正常，return;

如果Owner不為空，就是有上圖上面的類似于 bennu_index_benuprdapp02-1444748505181-f558155a-0 的文字，繼續看下面內容

第二步：查看消費端的程序代碼

一般的消費代碼是這樣的

看看自己的消費代碼里面，存不存在處理消息的時候出異常的情況

如果有，需要try-catch一下，其實不論有沒有異常，都用try-catch包一下最好，如下面代碼

return;

原因：如果在處理消息的時候有異常出現，又沒有進行處理，那么while循環就會跳出，線程會結束，所以不會再去取消息，就是消費停止了。

第三步：查看消費端的配置

消費代碼中一般以以下方式創建Consumer

消費端有一個配置，叫 fetch.message.max.bytes，默認是1M，此時如果有消息大于1M，會發生停止消費的情況。

此時，在配置中增加 props.put("fetch.message.max.bytes", "10 * 1024 * 1024"); 即可

return;

原因：目前Kafka集群配置的運行最大的消息大小是10M，如果客戶端配置的運行接收的消息是1M，跟Kafka服務端配置的不一致，

則消息大于1M的情況下，消費端就無法消費，導致一直卡在這一條消息，現象就是消費停止。

技術的提升是需要下苦工，需要堅持不懈的努力。就比如下面的分享的這些技術點，是否都學會并掌握了呢？如需要以下圖譜以及跟多提升架構技術的資源可加入我的粉絲Qqun：855801563。我花了將近一個月時間搜集整理了一套架構技術提升知識點講解以及一些面試題解析和答案免費分享給大家。助力各位程序員朋友突破自我提升技能，實現自己的目標。

總結

以上是生活随笔為你收集整理的jmeter线程数并发数区别_如何确定Kafka的分区数、key和consumer线程数、以及不消费问题解决...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。