java分布式流式處理組件Producer分區(qū)理論
前言
前面我們已經(jīng)對Producer發(fā)送原理做了一個比較詳細的說明，其中我們提到了分區(qū)器。其實從整體結(jié)構(gòu)上來講，分區(qū)器也是屬于一個非常重要的知識點，所以我們來專門對分區(qū)以及分區(qū)策略等內(nèi)容做一個介紹。
為什么需要分區(qū)
分區(qū)的作用
- 合理的使用存儲資源：把海量的數(shù)據(jù)按照分區(qū)切割成一小塊的數(shù)據(jù)存儲在多臺Broker上。此時能夠保證每臺服務(wù)器存儲資源能夠被充分利用到。而且小塊數(shù)據(jù)在尋址時間上更有優(yōu)勢~
如果將全部的數(shù)據(jù)存儲在一臺機器上，那么要對當(dāng)前數(shù)據(jù)做副本的時候，由于服務(wù)器資源配置不同，就有可能會出現(xiàn)副本數(shù)據(jù)存放失敗，從而增加數(shù)據(jù)丟失的可能性。
同時，如果單個文件過大，副本放置時間、內(nèi)容檢索時間都會極大的延長，從而導(dǎo)致Kafka性能降低。
- 負載均衡： 數(shù)據(jù)生產(chǎn)或消費期間，生產(chǎn)者已分區(qū)的單位發(fā)送數(shù)據(jù)，消費者分區(qū)的單位進行消費。 期間，各分區(qū)生產(chǎn)和消費數(shù)據(jù)互不影響，這樣能夠達到合理控制分區(qū)任務(wù)的程度，提高任務(wù)的并行度。從而達到負載均衡的效果。
剛才我們提到：生產(chǎn)者已分區(qū)為單位向Broker發(fā)送數(shù)據(jù)。那么問題來了：
- 生產(chǎn)者是怎么知道該向哪個分區(qū)發(fā)送數(shù)據(jù)呢？
這就是我們接下來要研究的分區(qū)策略。
分區(qū)策略
其實我們在上一篇文章中已經(jīng)見到了，看這里：
private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
// 如果在消息中指定了分區(qū)
if (record.partition() != null)
return record.partition();
if (partitioner != null) {
// 分區(qū)器通過計算得到分區(qū)
int customPartition = partitioner.partition(
record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
if (customPartition < 0) {
throw new IllegalArgumentException(String.format(
"The partitioner generated an invalid partition number: %d. Partition number should always be non-negative.", customPartition));
}
return customPartition;
}
// 通過序列化key計算分區(qū)
if (serializedKey != null && !partitionerIgnoreKeys) {
// hash the keyBytes to choose a partition
return BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size());
} else {
// 返回-1
return RecordMetadata.UNKNOWN_PARTITION;
}
}
下面的代碼可以說是整個分區(qū)器的核心部分，可以通過以下的步驟進行說明：
- 如果在生產(chǎn)消息的時候，已經(jīng)指定了需要發(fā)送的分區(qū)位置，那么就會直接使用已經(jīng)指定的份具體的位置，這樣子還節(jié)省了也不計算的時間
- 如果在生產(chǎn)者配置
Properties中指定了分區(qū)策略類，那么消息生產(chǎn)就會通過已經(jīng)指定的分區(qū)策略類進行分區(qū)計算 - 否則就會以
serializedKey作為參數(shù)，通過hash取模的方式計算。如果serializedKey == null，那么就會采用粘性分區(qū)的邏輯。 這在Kafka中屬于默認分區(qū)器。 - 如果以上情況都沒有包含，那么他就會直接返回-1。相當(dāng)于
ack=0的情況。
在Kafka中分區(qū)策略我們是可以自定義的。當(dāng)然Kafka也為我們內(nèi)置了三種分區(qū)策略類。 接下來我們挑個重點來介紹，來給我們自定義分區(qū)器做一個鋪墊～
我們已經(jīng)看到，DefaultPartitioner和UniformStickyPartitioner已經(jīng)被標(biāo)注為過期類，當(dāng)然也并不妨礙我們來了解一下。
DefaultPartitioner
在當(dāng)前版本中，如果沒有對partitioner.class進行配置，此時的分區(qū)策略就會采用當(dāng)前類作為默認分區(qū)策略類。
而以下是DefaultPartitioner策略類的核心實現(xiàn)方式，并且標(biāo)記部分的代碼實現(xiàn)其實就是UniformStickyPartitioner的計算邏輯
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster, int numPartitions) {
if (keyBytes == null) {
// 就是這段屬于UniformStickyPartitioner的實現(xiàn)邏輯
return stickyPartitionCache.partition(topic, cluster);
}
return BuiltInPartitioner.partitionForKey(keyBytes, numPartitions);
}
還有一段代碼讓我們來一起看看
public static int partitionForKey(final byte[] serializedKey, final int numPartitions) {
return Utils.toPositive(Utils.murmur2(serializedKey)) % numPartitions;
}
這段代碼不管有多復(fù)雜，調(diào)用方法有多少，但最終我們是能夠發(fā)現(xiàn)：
- 它的本質(zhì)其實是在對
序列化Key做哈希計算，然后通過hash值和分區(qū)數(shù)做取模運算，然后得到結(jié)果分區(qū)位置
這是一種比較重要的計算方式，但卻不是唯一的方式
---這是分割線---
接下來繼續(xù)，我們看看如果無法對序列化Key計算，會是怎么樣的計算邏輯？
我們先開始來看一下，是在哪個地方得到的serializedKey，并且什么情況下serializedKey會是NULL
看看下面的這個代碼眼熟不？
// 生產(chǎn)者生產(chǎn)消息對象
ProducerRecord<String, String> record = new ProducerRecord<>(
"newTopic001",
"data from " + KafkaQuickProducer.class.getName()
);
// KafkaProducer#doSend() // line994 serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
public class StringSerializer implements Serializer<String> {
// 省略。。。
@Override
public byte[] serialize(String topic, String data) {
if (data == null) {
return null;
} else {
return data.getBytes(encoding);
}
}
}
從上面的代碼來看，基本上能夠?qū)嶅N了：
- 當(dāng)在生成
ProducerRecord對象的時候，如果沒有對消息設(shè)置key參數(shù)，此時序列化之后的key就是個null - 那么當(dāng)序列化之后的Key為NULL之后，此時分區(qū)計算邏輯就會改變。
此時相當(dāng)于我們已經(jīng)進入到UniformStickyPartitioner的計算邏輯， 當(dāng)然了在我們使用的3.3版本中當(dāng)前類也已經(jīng)被標(biāo)注為過期
根據(jù)前面的說法，粘性分區(qū)主要解決了消息無Key的分區(qū)計算邏輯，那么粘性分區(qū)并不是說每次都使用同一個分區(qū)
它是通過一個大的Batch為單位，盡量將batch內(nèi)的消息固定在同一個分區(qū)內(nèi)，這樣在很大程度上能夠保證：
- 防止消息無規(guī)律的分散在不同的分區(qū)內(nèi)，降低分區(qū)傾斜
- 同時不需要每次進行分區(qū)計算，也降低了Producer的延遲
而實現(xiàn)方式是采用ConcurrentMap來進行緩存，感興趣的大家可以看看StickyPartitionCache的源碼
而當(dāng)Batch內(nèi)消息滿足發(fā)送條件被發(fā)送出去之后，才會開始再次計算下一個分區(qū)，為此在KafkaProducer中還專門調(diào)用了新的方法
partitioner.onNewBatch(topic, cluster, prevPartition);
public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
}
RoundRobinPartitioner
這是在當(dāng)前版本中唯一沒有被標(biāo)注的類，未來說不定會成為默認分區(qū)策略類，我們不看??，就瞄一眼
private int nextValue(String topic) {
AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> new AtomicInteger(0));
return counter.getAndIncrement();
}
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
int nextValue = nextValue(topic);
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
if (!availablePartitions.isEmpty()) {
int part = Utils.toPositive(nextValue) % availablePartitions.size();
return availablePartitions.get(part).partition();
} else {
// no partitions are available, give a non-available partition
return Utils.toPositive(nextValue) % numPartitions;
}
}
這個類的解釋，嗯。。你們看那個合適吧～
其實這個邏輯非常簡單：
- 通過
AtomicInteger.getAndIncrement()的方式將每次寫入平均分配到不同的分區(qū)中 - 不同與其他分區(qū)策略類，它不關(guān)心Key是否為NULL
我們先來做個小實驗吧： 將分區(qū)策略類修改為RoundRobinPartitioner，也方便后續(xù)自定義分區(qū)器的配置操作
config.setProperty(
ProducerConfig.PARTITIONER_CLASS_CONFIG,
"org.apache.kafka.clients.producer.RoundRobinPartitioner"
);
就這樣就能實現(xiàn)，看結(jié)果驗證～
中間穿插了一點小知識，那么接下來就會進入到我們最后一個環(huán)節(jié)：嘗試自定義分區(qū)器
自定義分區(qū)器
前面我們也提到過，相信大家沒有忘記partitioner.class這個配置
那么接下來就進入到重頭戲：自定義分區(qū)器實戰(zhàn)編碼環(huán)節(jié)。
public class CustomPartitioner implements Partitioner {
@Override
public void configure(Map<String, ?> configs) {
// nothing
}
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// 如果keyBytes == null
// 直接去0號位置
if (null == keyBytes) {
return 0;
}
// 已默認分區(qū)策略實現(xiàn)
int numPartitions = cluster.partitionsForTopic(topic).size();
return BuiltInPartitioner.partitionForKey(keyBytes, numPartitions);
}
@Override
public void close() {
// nothing
}
}
我們就先做的簡單一點，主要是想讓大家明白自定義分區(qū)器的實現(xiàn)：
- 如果沒有給定指定key，那么就默認全部去0號分區(qū)
- 否則就通過key做取模計算
當(dāng)自定義分區(qū)器實現(xiàn)完成之后，接下來我們就需要通過發(fā)送者進行驗證。當(dāng)然了，主要還是通過partitioner.class進行修改
// 給出關(guān)鍵代碼，其他的都是一樣的。就不贅述了~~~ config.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, "top.zopx.kafka.partitioner.CustomPartitioner");
通過執(zhí)行之后，我們來看看它的運行效果是否滿足我們的預(yù)期
另一種運行結(jié)果與默認分區(qū)器有Key的情況類似，這里就不再重復(fù)貼圖
代碼說明
本文全部代碼可進入Gitee中進行查看，更多精彩內(nèi)容敬請關(guān)注~
本次關(guān)于生產(chǎn)者分區(qū)器就介紹到這里，下期我們將推出針對Producer的生產(chǎn)優(yōu)化核心關(guān)注點，更多關(guān)于java分布式Producer分區(qū)的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
在進行項目debug時遇到了無法啟動的問題,項目一直處于正在啟動狀態(tài),但未出現(xiàn)任何報錯信息,分析原因可能是存在不合法的斷點位置,即斷點未打在方法內(nèi)部,解決方法是檢查所有斷點信息,并移除非法斷點,之后項目能夠正常啟動2023-02-02
Java請求調(diào)用參數(shù)格式為form-data類型的接口代碼示例
這篇文章主要給大家介紹了關(guān)于Java請求調(diào)用參數(shù)格式為form-data類型的接口的相關(guān)資料,文中給出了詳細的代碼示例,對大家的學(xué)習(xí)或者工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-08-08
這篇文章主要給大家介紹了關(guān)于LRU算法及Apache?LRUMap源碼解析的相關(guān)資料,文中通過實例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2021-11-11
使用@ConfigurationProperties實現(xiàn)類型安全的配置過程
這篇文章主要介紹了使用@ConfigurationProperties實現(xiàn)類型安全的配置過程，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-02-02
將本地SpringBoot項目發(fā)布到云服務(wù)器的方法
這篇文章主要介紹了如何將本地SpringBoot項目發(fā)布到云服務(wù)器,本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-12-12
這篇文章主要為大家詳細介紹了java生成指定范圍的隨機日期，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-12-12
java poi設(shè)置生成的word的圖片為上下型環(huán)繞以及其位置的實現(xiàn)
這篇文章主要介紹了java poi設(shè)置生成的word的圖片為上下型環(huán)繞以及其位置的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09
詳解Java對象轉(zhuǎn)換神器MapStruct庫的使用
在我們?nèi)粘ｉ_發(fā)的程序中，為了各層之間解耦，一般會定義不同的對象用來在不同層之間傳遞數(shù)據(jù)。當(dāng)在不同層之間傳輸數(shù)據(jù)時，不可避免地經(jīng)常需要將這些對象進行相互轉(zhuǎn)換。今天給大家介紹一個對象轉(zhuǎn)換工具MapStruct，代碼簡潔安全、性能高，強烈推薦2022-09-09
springboot 自定義配置Boolean屬性不生效的解決
這篇文章主要介紹了springboot 自定義配置Boolean屬性不生效的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-03-03
下載一些pdf格式的電子書沒有書簽,用JAVA寫了一個小工具，將特定格式的文本解析成為書簽，然后保存到pdf格式中2014-02-02