4、自定义分区器：实现Partitioner接口、基于Key哈希、基于业务字段、基于地理位置的分区策略

Kafka 默认的分区器（DefaultPartitioner）在无 Key 时采用轮询（Round-Robin），有 Key 时对 Key 的哈希值取模。但在读写分离、数据局部性、合规性要求下，默认策略往往导致分区倾斜或跨机房高延迟。本章深入自定义分区器的三种核心模式，并给出生产级实现。

4.1 自定义分区器接口与生命周期

所有自定义分区器必须实现 org.apache.kafka.clients.producer.Partitioner 接口，包含三个核心方法：

public class CustomPartitioner implements Partitioner {
    private String delimiter;

    @Override
    public void configure(Map<String, ?> configs) {
        // 从生产者配置中读取自定义参数
        this.delimiter = (String) configs.get("custom.partitioner.delimiter");
        if (delimiter == null) delimiter = ":";
    }

    @Override
    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();
        // 具体分区逻辑由子类实现
        return Math.abs(key.hashCode()) % numPartitions;
    }

    @Override
    public void close() { /* 清理资源 */ }
}

4.2 基于 Key 哈希的分区策略

这是最常用的策略，保证相同 Key 的消息落入同一分区，从而维持顺序。但需注意 Key 倾斜问题：当少数 Key 的数据量远大于其他 Key 时，会导致分区负载不均。

@Override
public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, Cluster cluster) {
    int numPartitions = cluster.partitionsForTopic(topic).size();
    if (keyBytes == null) {
        // 无 Key 时使用轮询
        return ThreadLocalRandom.current().nextInt(numPartitions);
    }
    // 使用 MurmurHash 替代 hashCode()，减少碰撞
    int hash = MurmurHash3.hash32(keyBytes);
    return Math.abs(hash) % numPartitions;
}

4.3 基于业务字段的分区策略

从消息 Value 中提取业务字段（如订单号、租户 ID、用户等级）决定分区。适用于 读写分离 场景：将高频写入的租户路由到更多分区，低频租户集中到少量分区。

@Override
public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, Cluster cluster) {
    int numPartitions = cluster.partitionsForTopic(topic).size();
    // 假设 Value 是 JSON 格式，提取 tenantId
    String jsonValue = new String(valueBytes, StandardCharsets.UTF_8);
    String tenantId = extractTenantId(jsonValue);
    // 租户与分区映射表（可从外部配置中心加载）
    int assignedPartition = tenantToPartitionMap.getOrDefault(tenantId, 0);
    return Math.min(assignedPartition, numPartitions - 1);
}

4.4 基于地理位置的分区策略

在跨地域部署的 Kafka 集群中，将数据路由到离消费者最近的 Broker 分区，显著降低消费延迟。核心思路：

• Broker 打标签： 每个 Broker 配置 broker.rack 或自定义标签（如 datacenter=shanghai）。
• 分区元数据： 通过 Cluster 对象获取每个分区的 Leader 副本所在 Broker 的机架信息。
• 消息携带地域： 从消息 Key 或 Value 中提取用户所在地域（如 IP 解析、GPS 坐标）。

@Override
public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, Cluster cluster) {
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    // 从消息中提取目标地域
    String targetRegion = extractRegionFromKey(key);
    // 遍历分区，找到 Leader 副本所在机架与目标地域匹配的分区
    for (PartitionInfo p : partitions) {
        Node leader = cluster.leaderFor(p.partition());
        if (leader != null && targetRegion.equals(leader.rack())) {
            return p.partition();
        }
    }
    // 无匹配时回退到哈希
    return Math.abs(keyBytes.hashCode()) % partitions.size();
}

4.5 分区策略选型对比

4.6 生产级配置与验证

在生产者端配置自定义分区器：

# 生产者配置文件
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
partitioner.class=com.example.CustomPartitioner
custom.partitioner.delimiter=: