kafka自定义分区实战
本文来说下kafka自定义分区相关的知识与内容,说下springboot整合kafka如何来实现自定义分区文章目录概述概述
·
本文来说下kafka自定义分区相关的知识与内容,同时说下springboot整合kafka如何来实现自定义分区
Kafka如何实现分区
Kafka集群是如何知道投递到哪个broker中
生产者在投递消息的时候传递key,然后根据key计算hash值存在到具体的broker中,如果是相同的key,最终投递消息都是同一个broker中。
默认分区策略
默认分区策略是:取正(bytearray生成32位hash值)%numpartitions。这个公式的结果是得到0-(numpartitions-1)间正整数的个数大致相等,也就是说kafka的默认分区策略是无论我们给定多少个分区,我们存放的数据基本上会平均的分到各个分区上。
package org.apache.kafka.clients.producer.internals;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;
public class DefaultPartitioner implements Partitioner {
private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();
public DefaultPartitioner() {
}
public void configure(Map<String, ?> configs) {
}
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
if (keyBytes == null) {
return this.stickyPartitionCache.partition(topic, cluster);
} else {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}
public void close() {
}
public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
this.stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
}
}
更多推荐
已为社区贡献7条内容
所有评论(0)