4. 自定义分区器（Custom Partitioner）开发：实现Partitioner接口，按业务字段路由

好，咱们接着聊分区策略。

前面讲了默认分区、轮询、粘性分区这些内置玩法。但说实话，在实际的金融系统里，这些往往不够用。为什么？因为金融数据太讲究“归属”了。

你想想看，一个订单数据，它必须落到同一个分区里，下游的消费者才能保证有序消费。如果按订单ID路由，那同一个订单的创建、支付、退款事件，必须去同一个分区。默认的轮询策略做不到这一点。这时候，就得我们自己动手，写一个自定义分区器。

4.1 什么时候需要自定义分区器？

我个人习惯是，只要业务上对“数据局部性”有强要求，就得上自定义分区器。比如：

• 按用户ID路由：同一个用户的所有行为日志，必须在一个分区，方便做用户画像。
• 按交易流水号路由：同一个交易的所有状态变更，必须有序。
• 按机构编码路由：不同机构的数据隔离，避免互相影响。

我在项目中遇到过这样一个场景：某支付系统，每天几千万笔交易。默认分区器把数据打散到各个分区，结果下游做对账的消费者，需要从多个分区拉取同一笔交易的数据，还得自己排序，性能惨不忍睹。后来改成按交易流水号哈希分区，问题直接解决。

4.2 实现Partitioner接口

Kafka提供的Partitioner接口，其实很简单。就两个核心方法：

• partition()：返回目标分区号。
• close()：资源清理，一般用不到。

我们重点看partition()方法。它的签名长这样：

public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, Cluster cluster);

参数里，key和value是我们发送消息时传的。说白了，路由的依据就在key里。如果你传的key是用户ID，那就在方法里对用户ID做哈希，然后模分区数。

嗯，这里要注意：keyBytes是序列化后的字节数组。如果key是null，那keyBytes也是null。所以代码里一定要判空。

4.3 实战：按业务字段路由

咱们直接写一个例子。假设我们有一个订单系统，消息的key是“订单ID”，格式是ORD_20250101_000001。我们希望同一个订单ID的所有消息，都去同一个分区。

代码实现如下：

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import java.util.List;
import java.util.Map;

public class OrderIdPartitioner implements Partitioner {

    @Override
    public void configure(Map<String, ?> configs) {
        // 可以在这里读取自定义配置，比如分区数
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 1. 获取该topic的所有分区信息
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        // 2. 如果key为null，使用默认策略（比如轮询）
        if (keyBytes == null || key == null) {
            // 这里简单处理，返回0。实际项目中建议用轮询或粘性分区
            return 0;
        }

        // 3. 对key做哈希，取绝对值，然后模分区数
        //    注意：hashCode可能为负数，需要处理
        int hash = Math.abs(key.hashCode());
        int partition = hash % numPartitions;

        // 4. 返回分区号
        return partition;
    }

    @Override
    public void close() {
        // 资源清理，本例中无需操作
    }
}

这段代码逻辑很清晰：

1. 拿到topic的分区总数。
2. 如果key是null，就落到0号分区（实际生产环境不建议这样，最好用默认分区器兜底）。
3. 对key做哈希，模分区数，得到目标分区。

4.4 注册并使用自定义分区器

写好了分区器，怎么用？很简单，在Producer配置里指定：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 指定自定义分区器
props.put("partitioner.class", "com.example.OrderIdPartitioner");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

然后发送消息时，把订单ID作为key传进去：

ProducerRecord<String, String> record = 
    new ProducerRecord<>("order-topic", "ORD_20250101_000001", "订单创建");
producer.send(record);

这样，所有订单ID为ORD_20250101_000001的消息，都会落到同一个分区。

4.5 更复杂的路由：从Value中提取字段

有时候，业务字段不在key里，而在value里。比如消息体是JSON，里面有个userId字段。这时候怎么办？

我的做法是：在partition()方法里，反序列化value，提取字段。但要注意，反序列化会消耗性能。所以一般建议：把路由字段放在key里。如果实在没办法，那就只能解析value了。

举个例子，假设value是JSON字符串：

{"userId": "U12345", "action": "login", "timestamp": 1700000000}

我们可以这样解析：

@Override
public int partition(String topic, Object key, byte[] keyBytes,
                     Object value, byte[] valueBytes, Cluster cluster) {
    if (valueBytes == null) {
        return 0;
    }
    try {
        String json = new String(valueBytes, StandardCharsets.UTF_8);
        // 用简单的字符串解析，或者用Jackson/Gson
        // 这里演示用字符串查找
        int start = json.indexOf("\"userId\":\"") + 10;
        int end = json.indexOf("\"", start);
        String userId = json.substring(start, end);
        int hash = Math.abs(userId.hashCode());
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        return hash % partitions.size();
    } catch (Exception e) {
        // 解析失败，落到默认分区
        return 0;
    }
}

4.6 自定义分区器的核心逻辑图

下面这张图，帮你理清整个流程：

流程很简单：Producer发消息 → 自定义分区器介入 → 根据业务字段算出目标分区 → 消息落入该分区。

4.7 几个重要的设计考量

写自定义分区器时，有几个点我建议你多想想：

考量点	说明	我的建议
分区数变化	如果topic扩容，分区数变了，哈希取模的结果会全部改变	尽量在初期规划好分区数，避免后期扩容导致数据重分布
key为null	如果key为null，自定义分区器需要兜底策略	可以用默认分区器，或者用粘性分区策略
性能开销	分区器在每次发送消息时都会调用	避免在partition()里做重量级操作，比如网络请求、复杂计算
数据倾斜	如果某些key的数据量特别大，会导致分区数据不均匀	可以考虑加盐（salted hash），或者用一致性哈希

4.8 总结一下

自定义分区器，说白了就是让你自己决定数据往哪走。实现Partitioner接口，重写partition()方法，把业务字段哈希到分区号上。就这么简单。

但简单归简单，坑也不少。key为null要处理，哈希负数要取绝对值，性能要关注，数据倾斜要预防。把这些点都照顾到了，你的自定义分区器才算合格。

嗯，下一节咱们聊聊分区策略的进阶玩法——如何动态调整分区，以及如何应对分区数变化带来的挑战。

---