[[322641]]

在使用Apache Kafka生產(chǎn)和消費(fèi)消息的時(shí)候，肯定是希望能夠?qū)?shù)據(jù)均勻地分配到所有服務(wù)器上。

比如很多公司使用Kafka收集應(yīng)用服務(wù)器的日志數(shù)據(jù)，這種數(shù)據(jù)都是很多的，特別是對于那種大批量機(jī)器組成的集群環(huán)境，每分鐘產(chǎn)生的日志量都能以GB數(shù)，因此如何將這么大的數(shù)據(jù)量均勻地分配到Kafka的各個Broker上，就成為一個非常重要的問題。

為什么分區(qū)?

Topic的概念，它是承載真實(shí)數(shù)據(jù)的邏輯容器，而在主題之下還分為若干個分區(qū)，也就是說Kafka的消息組織方式實(shí)際上是三級結(jié)構(gòu)：主題-分區(qū)-消息。主題下的每條消息只會保存在某一個分區(qū)中，而不會在多個分區(qū)中被保存多份。官網(wǎng)上的這張圖非常清晰地展示了。

Kafka的三級結(jié)構(gòu)，如下所示：

看到了這張圖，我有幾個問題，為什么Kafka要做這樣的設(shè)計(jì)?為什么使用分區(qū)而不是直接使用多個Topic呢?

分區(qū)的作用

其實(shí)，分區(qū)的作用就是提供負(fù)載均衡的能力，或者說對數(shù)據(jù)進(jìn)行分區(qū)的主要原因，就是為了實(shí)現(xiàn)系統(tǒng)的高伸縮性(Scalability)

不同的分區(qū)能夠被放置到不同節(jié)點(diǎn)的機(jī)器上，而數(shù)據(jù)的讀寫操作也都是針對分區(qū)這個粒度而進(jìn)行的，這樣每個節(jié)點(diǎn)的機(jī)器都能獨(dú)立地執(zhí)行各自分區(qū)的讀寫請求處理，并且，我們還可以通過添加新的節(jié)點(diǎn)機(jī)器來增加整體系統(tǒng)的吞吐量

實(shí)際上分區(qū)的概念以及分區(qū)數(shù)據(jù)庫早在1980年就已經(jīng)有大牛們在做了，比如那時(shí)候有個叫Teradata的數(shù)據(jù)庫就引入了分區(qū)的概念

在不同的分布式系統(tǒng)對分區(qū)的叫法也不盡相同：比如在Kafka中叫分區(qū)，在MongoDB和Elasticsearch中就叫分片Shard，而在HBase中則叫Region，在Cassandra中又被稱作vnode

從表面看起來，它們實(shí)現(xiàn)原理可能不盡相同，但對底層分區(qū)(Partitioning)的整體思想?yún)s從未改變

除了提供負(fù)載均衡這種最核心的功能之外，利用分區(qū)也可以實(shí)現(xiàn)其他一些業(yè)務(wù)級別的需求，比如實(shí)現(xiàn)業(yè)務(wù)級別的消息順序的問題

Kafka中的分區(qū)策略

Kafka中的分區(qū)策略，就是決定生產(chǎn)者將消息發(fā)送到哪個分區(qū)的算法

Kafka提供了默認(rèn)的分區(qū)策略，同時(shí)，也支持自定義分區(qū)策略

• 默認(rèn)分區(qū)策略
• 自定義分區(qū)策略

默認(rèn)分區(qū)策略

• 輪詢策略(Round-robin)
• 隨機(jī)策略(Randomness)(已過時(shí))
• 消息鍵策略(Key-ordering)
• 地理分區(qū)策略

輪詢策略

也稱Round-robin策略，即順序分配

比如一個主題下有3個分區(qū)，那么第一條消息被發(fā)送到分區(qū)0，第二條被發(fā)送到分區(qū)1，第三條被發(fā)送到分區(qū)2，以此類推。當(dāng)生產(chǎn)第4條消息時(shí)又會重新開始，即將其分配到分區(qū)0，如下圖所示

如果你未指定partitioner.class參數(shù)，那么你的生產(chǎn)者程序會按照輪詢的方式在Topic的所有分區(qū)間均勻地“存放”消息

輪詢策略有非常優(yōu)秀的負(fù)載均衡表現(xiàn)，它總是能保證消息最大限度地被平均分配到所有分區(qū)上，默認(rèn)情況下它是最合理的分區(qū)策略，也是我們最常用的分區(qū)策略之一

隨機(jī)策略

也稱Randomness策略，所謂隨機(jī)就是我們隨意地將消息放置到任意一個分區(qū)上，如下圖所示

如果要實(shí)現(xiàn)隨機(jī)策略版的partition方法，很簡單，只需要兩行代碼即可：

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
return ThreadLocalRandom.current().nextInt(partitions.size()); 

先計(jì)算出該Topic總的分區(qū)數(shù)，然后隨機(jī)地返回一個小于它的正整數(shù)

本質(zhì)上看隨機(jī)策略也是力求將數(shù)據(jù)均勻地打散到各個分區(qū)，但從實(shí)際表現(xiàn)來看，它要遜于輪詢策略，所以如果追求數(shù)據(jù)的均勻分布，還是使用輪詢策略比較好

事實(shí)上，隨機(jī)策略是老版本生產(chǎn)者使用的分區(qū)策略，在新版本中已經(jīng)改為輪詢了

消息鍵策略

也稱Key-ordering策略，Kafka允許為每條消息定義消息鍵，簡稱為Key

這個Key的作用非常大，它可以是一個有著明確業(yè)務(wù)含義的字符串，比如客戶代碼、部門編號或是業(yè)務(wù)ID等;也可以用來表征消息元數(shù)據(jù)

特別是在Kafka不支持時(shí)間戳的年代，在一些場景中，工程師們都是直接將消息創(chuàng)建時(shí)間封裝進(jìn)Key里面的

一旦消息被定義了Key，那么你就可以保證同一個Key的所有消息都進(jìn)入到相同的分區(qū)里面，由于每個分區(qū)下的消息處理都是有順序的，故這個策略被稱為按消息鍵策略，如下圖所示

實(shí)現(xiàn)這個策略的partition方法同樣簡單，只需要下面兩行代碼即可：

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
return Math.abs(key.hashCode()) % partitions.size(); 

先計(jì)算出該Topic總的分區(qū)數(shù)，然后計(jì)算出key的hashCode與分區(qū)數(shù)取模的絕對值

Kafka在默認(rèn)分區(qū)策略的選擇：如果指定了Key，那么默認(rèn)實(shí)現(xiàn)按消息鍵策略;如果沒有指定Key，則使用輪詢策略

地理分區(qū)策略

上面這幾種分區(qū)策略都是比較基礎(chǔ)的策略，其實(shí)還有一種比較常見的，即所謂的基于地理位置的分區(qū)策略

當(dāng)然這種策略一般只針對那些大規(guī)模的Kafka集群，特別是跨城市、跨國家甚至是跨大洲的集群

自定義分區(qū)策略

說完了默認(rèn)分區(qū)，來說說自定義分區(qū)

Kafka中如果要自定義分區(qū)策略，你需要顯式地配置生產(chǎn)者端的參數(shù)partitioner.class

這個參數(shù)該怎么設(shè)定呢?方法很簡單，在編寫生產(chǎn)者程序時(shí)，你可以編寫一個具體的類實(shí)現(xiàn)org.apache.kafka.clients.producer.Partitioner接口

這個接口也很簡單，只定義了兩個方法：partition()和close()，通常你只需要實(shí)現(xiàn)最重要的partition方法，代碼如下所示

/** 
     * Compute the partition for the given record. 
     * 
     * @param topic The topic name 
     * @param key The key to partition on (or null if no key) 
     * @param keyBytes The serialized key to partition on( or null if no key) 
     * @param value The value to partition on or null 
     * @param valueBytes The serialized value to partition on or null 
     * @param cluster The current cluster metadata 
     */ 
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster); 
 
    /** 
     * This is called when partitioner is closed. 
     */ 
    public void close(); 

這里的topic、key、keyBytes、value和valueBytes都屬于消息數(shù)據(jù)，cluster則是集群信息(比如當(dāng)前Kafka集群共有多少主題、多少Broker等)

Kafka給你這么多信息，就是希望讓你能夠充分地利用這些信息對消息進(jìn)行分區(qū)，計(jì)算出它要被發(fā)送到哪個分區(qū)中

只要你自己的實(shí)現(xiàn)類定義好了partition方法，同時(shí)設(shè)置partitioner.class參數(shù)為你自己實(shí)現(xiàn)類的Full Qualified Name，那么生產(chǎn)者程序就會按照你的代碼邏輯對消息進(jìn)行分區(qū)

結(jié)語

今天學(xué)習(xí)了Kafka生產(chǎn)者消息分區(qū)的機(jī)制以及常見的幾種分區(qū)策略

分區(qū)是實(shí)現(xiàn)負(fù)載均衡以及高吞吐量的關(guān)鍵，故在生產(chǎn)者這一端就要仔細(xì)盤算合適的分區(qū)策略，避免造成消息數(shù)據(jù)的傾斜，使得某些分區(qū)成為性能瓶頸，這樣極易引發(fā)下游數(shù)據(jù)消費(fèi)的性能下降