本文来说下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);
    }
}

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐